Support schema evolution / schema overwrite in DeltaLake MERGE

As far as I can tell, schema evolution / schema overwrite in DeltaLake MERGE is not currently supported. The below pyspark code illustrates my issue (Spark 2.4.4, Scala 2.11, DeltaLake 0.3.0):

schema1 = StructType([ StructField("id", IntegerType()),
                       StructField("col1", StringType()),  
                       StructField("col2", IntegerType())
                      ])
# Define DFs
DF1 = spark.createDataFrame([ [1, "a", 1],
                              [2, "a", 5], 
                              [3, "b", 5]], schema=schema1)

schema2 = StructType([
  StructField("id", IntegerType()),
  StructField("col1", StringType()),  
  StructField("col2", DoubleType()),
  StructField("col3", StringType())  
])

myDF2 =spark.createDataFrame([ [1, "b", 4.0, "a"],
                               [2, "c", 5.0, "b"], 
                               [4, "b", 5.0, "c"]], schema=schema2) 

# Write first DF as a DeltaTable
DF2.write.format("delta") \
          .mode("overwrite") \
          .option("overwriteSchema", "true") \
          .save("/tmp/delta/testtable")

# DeltaLake merge via py4j
jvm_for_table = spark._jvm.io.delta.tables.DeltaTable \
                             .forPath(spark._jsparkSession, "/tmp/delta/testtable")

# workaround to avoid using Python keyword 'as' in JVM method call 
# (Python interpreter will throw an error.)
jvm_for_table_as=getattr(jvm_for_table, 'as')("testtable")

# update existing rows and insert new
jvm_for_table_as.merge(getattr(DF2._jdf, 'as')("updates"), \
"testtable.id = updates.id").whenMatched().updateAll().execute()

# Read merged table and show
spark.read.format("delta").load("/tmp/delta/testtable").show()

This quietly outputs:

+---+----+------+
| id |col1| col2|
+---+----+------+
|  1|   a  | 1  |
|  2|   a  | 5  |
|  3|   b  | 5  |
+---+----+------+

Would be great, if by option (.option("overwriteSchema", "true") .option("mergeSchema", "true") ), to support schemaevolution to get instead:

+---+----+------+------+
| id |col1| col2| col3 |
+---+----+------+------+
|  1|   b  | 4.0 |  a  |
|  2|   c  | 5.0 |  b  |
|  3|   b  | 5.0 |None |
|  4|   b  | 5.0 |  c  |
+---+----+------+------+

and a new schema

StructType([
  StructField("id", IntegerType()),
  StructField("col1", StringType()),  
  StructField("col2", DoubleType()),
  StructField("col3", StringType())  
])

Any update on this issue?

Not yet. It is something we want to do eventually, but we're not sure we're going to get to it this quarter. It is on our roadmap though.

I have implemented a solution to do schema evolution in the interim. In my use case I wanted to have the sink represent a superset of columns, i.e. everything that existed at any given stage in the source.

Hope that helps in the interim!

P.s. Don't judge my Python coding skills, I'm self-taught!

@gerardwolf

@gerardwolf this is great! thank you. it will be super helpful if you paste the code directly in comments in markdown format rather than screenshots. would be much easier for others to copy and use the code.

I was a little lazy, source code zipped and attached.

DeltaSchemaMerge.zip

will this be supported in the next version? we have an issue with this at the moment.

we are hoping to add this for 0.6.0

@gerardwolf

I am trying to use this workaround of yours @gerardwolf, and I am wondering about what the FILTER=reduce does in the SQL query, could you please explain it a bit for me? :)

@Lytiker

@Lytiker. the "reduce" is a variable I defined earlier in the notebook which can contain a WHERE clause. I use the same code to load all my source objects, so this allows me to populate the "reduce" variable with a clause relevant to a specific objects filter criteria. It allows me to MERGE INTO the target delta table using the same filter condition I have on the source query. Basically so I compare apples and apples in the source and sink objects. Hope that makes sense.

@gerardwolf

ah, thank you @gerardwolf , so do I understand you correctly, if I say this is where you would handle the case of incorporating rows for id 1,2 and 4 into DF1 of the example on top of this thread?

@samkutty94

@samkutty94 yes it is. all APIs have same functionality and semantics.

@tdas

@tdas
You mentioned that updateAll & insertAll covers most use cases, does it also cover scenarios where SCD Type 2 needs to be implemented on the same table that needs to have schema evolution?

For SCD type 2 I want to be able to 'retire' an existing / matching row and add another one with appropriate flags / dates, but I can't do that If i'm forced to use updateAll & insertAll

It seems I'm going to have to choose which feature I can have in my etl process using Delta Lake or is there an alternative approach which isn't documented? Any pointers?

Thanks

@tdas

@tdas what is the depth with which schema evolution works while merging?

Automatic schema evolution does not work while merging in the following case.

import json
d1 = {'a':'b','b':{'c':{'1':1}}}
d2 = {'a':'s','b':{'c':{'1':2,'2':2}}}
d3 = {'a':'v','b':{'c':{'1':4}}}

df1 = spark.read.json(spark.sparkContext.parallelize([json.dumps(d1)]))

#passes
df1.write.saveAsTable('test_table4',format='delta',mode='overwrite', path=f"hdfs://hdmaster:9000/dest/test_table4")


df2 = spark.read.json(spark.sparkContext.parallelize([json.dumps(d2)]))
df2.createOrReplaceTempView('updates')

query = """
MERGE INTO test_table4 existing_records 
        USING updates updates 
        ON existing_records.a=updates.a
        WHEN MATCHED THEN UPDATE SET * 
        WHEN NOT MATCHED THEN INSERT *
"""
spark.sql("set spark.databricks.delta.schema.autoMerge.enabled=true")
spark.sql(query) #passes



df3 = spark.read.json(spark.sparkContext.parallelize([json.dumps(d3)]))

df3.createOrReplaceTempView('updates')
query = """
MERGE INTO test_table4 existing_records 
        USING updates updates 
        ON existing_records.a=updates.a
        WHEN MATCHED THEN UPDATE SET * 
        WHEN NOT MATCHED THEN INSERT *
"""
spark.sql("set spark.databricks.delta.schema.autoMerge.enabled=true")
spark.sql(query) #FAILS #FAILS

This looks like failing when depth is more than 2 and incoming df has columns missing.
Is this intentionally like this?
This is handled perfectly with option("mergeSchema", "true") if want to append. But I want to UPSERT the data. But Merge is not able to handle this schema change

Using Delta Lake version 0.8.0

what is the error?

@tdas

@tdas thanks for the quick response.
This is the error I get when running the above code. Cannot cast struct<1:bigint> to struct<1:bigint,2:bigint>. All nested columns must match.;

Noting that if a similar schema change(missing of column in incoming data) happens on the upper levels(I think till depth 2) of df, it works fine.

can you give the full stack trace?

@tdas

@tdas sure, Below is the full stack trace of the error:

---------------------------------------------------------------------------
AnalysisException                         Traceback (most recent call last)
<ipython-input-1-a5bf96348d93> in <module>
     36 """
     37 spark.sql("set spark.databricks.delta.schema.autoMerge.enabled=true")
---> 38 spark.sql(query) #FAILS #FAILS

/usr/local/spark/python/pyspark/sql/session.py in sql(self, sqlQuery)
    647         [Row(f1=1, f2=u'row1'), Row(f1=2, f2=u'row2'), Row(f1=3, f2=u'row3')]
    648         """
--> 649         return DataFrame(self._jsparkSession.sql(sqlQuery), self._wrapped)
    650 
    651     @since(2.0)

/usr/local/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1303         answer = self.gateway_client.send_command(command)
   1304         return_value = get_return_value(
-> 1305             answer, self.gateway_client, self.target_id, self.name)
   1306 
   1307         for temp_arg in temp_args:

/usr/local/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
    132                 # Hide where the exception came from that shows a non-Pythonic
    133                 # JVM exception message.
--> 134                 raise_from(converted)
    135             else:
    136                 raise

/usr/local/spark/python/pyspark/sql/utils.py in raise_from(e)

AnalysisException: Cannot cast struct<1:bigint> to struct<1:bigint,2:bigint>. All nested columns must match.;

is there any more of the stack trace? the java part of the stack trace after the last line you have shown?

@tdas

@tdas no. that is all it gives.

I am facing issue while merging the schema with existing Delta table, please find below scenario -
At first iteration of processing the Json file -
Delta table created by inferring schema of Json file and so it got a column (For say Column Name is Col1) with DataType as StringType where value was NULL initially.

Second iteration, while appending another Json file with this Delta table -
Col1 got some StructType Data in the Json file

While appending with option mergeSchema = true, it is throwing error something like below -
Failed to merge fields Col1 and Col1. Failed to merge incompatible data types StringType and StructType

@divasgupta

@divasgupta could you please share the code and the data you are merging I might be able to help

@Rohit25negi

@Rohit25negi I am facing the same problem, how did you solve it?

@divasgupta

I am facing a similar issue as @divasgupta,
I have columns c1 in my first iteration, while c1 is dropped in the second iteration.

The error I get is The specified schema does not match the existing schema

[delta-io#125] Fix a bug in CloseableParquetDataIterator where a sing…

does this work now? I can't seem to get it working, it just drops the new columns

Navigation Menu

Repository navigation

Support schema evolution / schema overwrite in DeltaLake MERGE #170

12 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Support schema evolution / schema overwrite in DeltaLake MERGE #170

Description

Activity

subashsivaji commented on Nov 13, 2019

mukulmurthy commented on Nov 13, 2019

gerardwolf commented on Dec 11, 2019

tdas commented on Dec 11, 2019

gerardwolf commented on Dec 12, 2019

kkr78 commented on Feb 21, 2020

tdas commented on Feb 24, 2020

Lytiker commented on Mar 30, 2020

gerardwolf commented on Mar 30, 2020

Lytiker commented on Mar 31, 2020

12 remaining items

tdas commented on Oct 1, 2020

thenorthnomad commented on Feb 1, 2021

Rohit25negi commented on Feb 23, 2021

tdas commented on Feb 23, 2021

Rohit25negi commented on Feb 24, 2021

tdas commented on Feb 24, 2021

Rohit25negi commented on Feb 25, 2021

tdas commented on Feb 25, 2021

Rohit25negi commented on Feb 25, 2021

divasgupta commented on Mar 4, 2021

Rohit25negi commented on Mar 5, 2021

francescomucio commented on Nov 29, 2021

rachiitgupta commented on Feb 3, 2022

TheKnightCoder commented on May 11, 2024

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions