Load only struct from map's value from an avro file into a Spark Dataframe

user_7154914 0
Viewed 93

Using PySpark, I need to load "Properties" object (map's value) from an avro file into its own Spark dataframe. Such that, "Properties" from my avro file will become a dataframe with its elements and values as columns and rows. Hence, struggling to find some clear examples accomplishing that.

Schema of the file:

root
 |-- SequenceNumber: long (nullable = true)
 |-- Offset: string (nullable = true)
 |-- EnqueuedTimeUtc: string (nullable = true)
 |-- SystemProperties: map (nullable = true)
 |    |-- key: string
 |    |-- value: struct (valueContainsNull = true)
 |    |    |-- member0: long (nullable = true)
 |    |    |-- member1: double (nullable = true)
 |    |    |-- member2: string (nullable = true)
 |    |    |-- member3: binary (nullable = true)
 |-- Properties: map (nullable = true)
 |    |-- key: string
 |    |-- value: struct (valueContainsNull = true)
 |    |    |-- member0: long (nullable = true)
 |    |    |-- member1: double (nullable = true)
 |    |    |-- member2: string (nullable = true)
 |    |    |-- member3: binary (nullable = true)
 |-- Body: binary (nullable = true)

The resulting "Properties" dataframe loaded from the above avro file needs to be like this:

member0 member1 member2 member3
value value value value
pyspark apache-spark-sql apache-spark struct dictionary
1 Answers

map_values is your friend.

Collection function: Returns an unordered array containing the values of the map.
New in version 2.3.0.

df_properties = df.select((F.map_values(F.col('Properties'))[0]).alias('vals')).select('vals.*')

Full example:

df = spark.createDataFrame(
    [('a', 20, 4.5, 'r', b'8')],
    ['key', 'member0', 'member1', 'member2', 'member3'])
df = df.select(F.create_map('key', F.struct('member0', 'member1', 'member2', 'member3')).alias('Properties'))
df.printSchema()
# root
#  |-- Properties: map (nullable = false)
#  |    |-- key: string
#  |    |-- value: struct (valueContainsNull = false)
#  |    |    |-- member0: long (nullable = true)
#  |    |    |-- member1: double (nullable = true)
#  |    |    |-- member2: string (nullable = true)
#  |    |    |-- member3: binary (nullable = true)

df_properties = df.select((F.map_values(F.col('Properties'))[0]).alias('vals')).select('vals.*')

df_properties.show()
# +-------+-------+-------+-------+
# |member0|member1|member2|member3|
# +-------+-------+-------+-------+
# |     20|    4.5|      r|   [38]|
# +-------+-------+-------+-------+

df_properties.printSchema()
# root
#  |-- member0: long (nullable = true)
#  |-- member1: double (nullable = true)
#  |-- member2: string (nullable = true)
#  |-- member3: binary (nullable = true)
user_27535010
Related