PySpark: Length of object does not match with length of fields - creating new schema

Viewed 1705

I am currently working with AWS Glue and PySpark. I am trying to manually create my Spark schema and apply it to the dataframe to fix some issues I have with some columns.

My current problem is with columns with struct type. My data only has 6 fields provided in the JSON read via Glue but it could have 8 fields in the future. If I build my schema with the 6 fields I receive so far, it works fine but if I build the schema with the 8 fields I am supposed to get, I get the following error: ValueError: field name_struct: Length of object (6) does not match with length of fields (8).

What would be the best way to apply my updated schema? I think that I should apply a function to my struct column and create the missing keys with null values to have the expected 8 keys in my data. Any idea on how to do that? If I do that, I think that for all my struct type columns, I should parse the schema, compare the keys between the data in the column (asDict()?) and the schema and finally add the missing ones. I have the feeling that there is a better way to do with Spark or maybe some parameters I haven't found yet to automatically fix that when creating the dataframe?

import findspark

findspark.init()

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType, IntegerType, StructType, StructField

spark = SparkSession.builder.master("local[1]") \
                .appName('test_schema') \
                .getOrCreate()
                
data = [
    (32, ("James", "Smith"), ("street", 42)),
    (18, ("Nina", "Smith"), ("street 2", 12))
]

schema = StructType([
    StructField('age', IntegerType(), True),
    StructField('name_struct', StructType([
        StructField('first_name', StringType(), True),
        StructField('last_name', StringType(), True)
    ])),
    StructField('address_struct', StructType([
        StructField('street', StringType(), True),
        StructField('number', IntegerType(), True)
    ]))
])

df = spark.createDataFrame(data=data, schema=schema)
df.printSchema()
df.show()

schema2 = StructType([
    StructField('age', IntegerType(), True),
    StructField('name_struct', StructType([
        StructField('first_name', StringType(), True),
        StructField('last_name', StringType(), True),
        StructField('middle_name', StringType(), True)
    ])),
    StructField('address_struct', StructType([
        StructField('street', StringType(), True),
        StructField('number', IntegerType(), True),
        StructField('zip_code', IntegerType(), True)
    ]))
])

# fix data here?

df2 = spark.createDataFrame(data=data, schema=schema2)
# error here:  ValueError: field name_struct: Length of object (2) does not match with length of fields (3)
df2.printSchema()
df2.show()

Thanks!

0 Answers
Related