How to Extract Columns From BinaryType Using pySpark Databricks?

Question

Issue: Extracting columns from dataframe's binary type column. The data frame was loaded from blob storage account of azure.

Environment:

• Databricks 5.4 (includes Apache Spark 2.4.3)
• Python 3.5.2

Process:

1. Get data from avro files
2. Extract useful information and write back more user friendly version to parquet

Avro schema:

    SequenceNumber:long
    Offset:string
    EnqueuedTimeUtc:string
    SystemProperties:map
        key:string
        value:struct
            member0:long
            member1:double
            member2:string
            member3:binary
    Properties:map
        key:string
        value:struct
            member0:long
            member1:double
            member2:string
            member3:binary
    Body:binary

I struggle getting data from Body:binary. I managed to convert column to string using code snippet below

df = df.withColumn("Body", col("Body").cast("string"))

I managed to extract a list of columns in the body column using code below:

        #body string looks like json
        dfBody = df.select(df.Body)
        jsonList = (dfBody.collect())
        jsonString = jsonList[0][0]
        columns = []
        data = json.loads(jsonString)

        for key, value in data.items():
            columns.append(key)

        columns.sort()
        print(columns) 

The list has interesting columns such as ID, Status, Name.

Question: How do I add ID column that sits in body binary column and add to my current dataframe. In general, I want to flatten binary column. Binary column might also have arrays.

Answer 1

You don't want to collect the dataframe. Instead you should be able to cast and flatten the body field. From the looks of it you are using avro captures from Event Hubs. This is code that I use to handle this:

from pyspark.sql.types import StringType, IntegerType, StructType, StructField
from pyspark.sql.functions import from_json, col

# Create a schema that describes the Body field
sourceSchema = StructType([
        StructField("Attribute1", StringType(), False),
        StructField("Attribute2", StringType(), True),
        StructField("Attribute3", StringType(), True),
        StructField("Attribute4", IntegerType(), True)])


# Convert Body to String and then Json applying the schema
df = df.withColumn("Body", col("Body").cast("string"))
jsonOptions = {"dateFormat" : "yyyy-MM-dd HH:mm:ss.SSS"}
df = df.withColumn("Body", from_json(df.Body, sourceSchema, jsonOptions))

# Flatten Body
for c in df.schema['Body'].dataType:
    df = df.withColumn(c.name, col("Body." + c.name))

I think the key bit you need is the from_json function.