Pyspark delta lake json datatype evolution issue ( merge incompatible exception )

user_5311367 0
Viewed 3762

I am working on pyspark (3.x) and delta lake. I am facing some challenges w.r.t to datatypes. We are receiving data as JSON data type, we are doing some flattening on the JSON datasets and saving it as delta tables with options as "mergeSchema" as true as shown below. We are not imposing any schema on the table.

df.write\
    .format("delta")\
    .partitionBy("country","city")\
    .option("mergeSchema","true")\
    .mode("append")\
    .save(delta_path)\

The problem we are facing is- the data type of JSON fields gets change very often,for example In delta table "field_1" is getting stored with datatype as StringType but the datatype for 'field_1' for new JSON is coming as LongType. Due to this we are getting merge incompatible exception.

ERROR : Failed to merge fields 'field_1' and 'field_1'. Failed to merge incompatible data types StringType and LongType

How to handle such datatype evolution in delta tables, I dont want to handle datatype changes at a field level because we have more than 300+ fields coming as part of json.

delta-lake pyspark jsonschema json
3 Answers

According to the article Diving Into Delta Lake: Schema Enforcement & Evolution the option mergeSchema=true can handle the following scenarios:

  • Adding new columns (this is the most common scenario)
  • Changing of data types from NullType -> any other type, or upcasts from ByteType -> ShortType -> IntegerType

The article also gives a hint on what can be done in your case:

"Other changes, which are not eligible for schema evolution, require that the schema and data are overwritten by adding .option("overwriteSchema", "true"). For example, in the case where the column “Foo” was originally an integer data type and the new schema would be a string data type, then all of the Parquet (data) files would need to be re-written. Those changes include:"

  • Dropping a column
  • Changing an existing column’s data type (in place)
  • Renaming column names that differ only by case (e.g. “Foo” and “foo”)
user_122089100
user_169554490

In order get my issue resolved, I have written a new function that essentially merges schema of the delta table (if delta table exist) and JSON schema.

At a high level, I have created a new schema- this new schema is essentially a combination of common columns from delta lake table and new columns from JSON fields, by creating this new schema I recreate a data frame by applying this new schema. This has solved my issue.

def get_merged_schema(delta_table_schema, json_data_schema):
    
    print('str(len(delta_table_schema.fields)) -> ' + str(len(delta_table_schema.fields)))
    print('str(len(json_data_schema.fields)) -> '+ str(len(json_data_schema.fields)))
    
    no_commom_elements=False
    no_new_elements=False
    import numpy as np
    struct_field_array=[]
    if len(set(delta_table_schema.names).intersection(set(json_data_schema.names))) > 0:
        common_col=set(delta_table_schema.names).intersection(set(json_data_schema.names))
        print('common_col len: -> '+ str(len(common_col)))
        for name in common_col:
            for f in delta_table_schema.fields:
              if(f.name == name):
                  struct_field_array.append(StructField(f.name, f.dataType, f.nullable))
    else:
        no_commom_elements=True
        print("no common elements")

    if len(np.setdiff1d(json_data_schema.names,delta_table_schema.names)) > 0:
        diff_list = np.setdiff1d(json_data_schema.names,delta_table_schema.names)
        print('diff_list len: -> '+ str(len(diff_list)))
        for name in diff_list:
            for f in json_data_schema.fields:
                if(f.name == name):
                    struct_field_array.append(StructField(f.name, f.dataType, f.nullable))
    else:
        no_new_elements=True
        print("no new elements")
      
    print('len(StructType(struct_field_array)) -> '+str(len(StructType(struct_field_array))))
    df=spark.createDataFrame(spark.sparkContext.emptyRDD(),StructType(struct_field_array))
    if no_commom_elements and no_new_elements: 
      return StructType(None) 
    else: 
      return df.select(sorted(df.columns)).schema
user_53113670
Related