topic Re: Tring to create incremental pipeline but fails when I try to use outputMode "update" in Data Engineering

Tring to create incremental pipeline but fails when I try to use outputMode "update"

BorislavBlagoev — Wed, 19 Jan 2022 17:27:17 GMT

def upsertToDelta(microBatchOutputDF, batchId): 
  
  microBatchOutputDF.createOrReplaceTempView("updates")
 
  microBatchOutputDF._jdf.sparkSession().sql("""
    MERGE INTO old o
    USING updates u
    ON u.id = o.id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)
 
stream_new_df = spark.readStream.format("delta").load(new_data_frame_path)
stream_old_df = spark.readStream.format("delta").load(old_data_frame_path)
 
stream_old_df.createOrReplaceTempView("old")
 
stream_new_df.writeStream.format("delta") \
            .option("checkpointLocation", "") \
            .option("mergeSchema", "true") \
            .option("path", "") \
            .foreachBatch(upsertToDelta) \
            .trigger(once=True) \
            .outputMode("update") \
            .table("")

I'm trying to execute this code but I get the following error:

Data source com.databricks.sql.transaction.tahoe.sources.DeltaDataSource does not support Update output mode

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

Hubert-Dudek — Wed, 19 Jan 2022 17:52:57 GMT

Delta table/file version is too old. Please try to upgrade it as described here https://docs.microsoft.com/en-us/azure/databricks/delta/versioning

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

BorislavBlagoev — Wed, 19 Jan 2022 17:56:18 GMT

Which is the latest version?

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

BorislavBlagoev — Wed, 19 Jan 2022 18:20:55 GMT

@Hubert Dudek I get the same error

AnalysisException: Data source com.databricks.sql.transaction.tahoe.sources.DeltaDataSource does not support Update output mode

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

BorislavBlagoev — Wed, 19 Jan 2022 18:32:00 GMT

I tried with the both ways

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

Hubert-Dudek — Thu, 20 Jan 2022 10:38:05 GMT

Did it work? Databricks runtime is also imported as older one (like one used by data factory)

I think you can also refactor code a bit to use .start() in last line not .table() and change a bit def upsertToDelta to just use something like that (it is in scala but similar logic for python) https://docs.databricks.com/_static/notebooks/merge-in-streaming.html

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

BorislavBlagoev — Thu, 20 Jan 2022 10:42:04 GMT

@Hubert Dudek The runtime version is 9.1LTS. And I want to use the `.table()` because I want to have a table in my metastore/catalog

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

BorislavBlagoev — Thu, 20 Jan 2022 11:06:08 GMT

@Hubert Dudek I also tried with 10.2 runtime and with toTable() but it's the same

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

Hubert-Dudek — Thu, 20 Jan 2022 11:54:19 GMT

to have table in metastore just register your delta location there using seperate sql script (it is enough to do that one time):

%sql
CREATE TABLE IF NOT EXISTS your_db.your_table
( 
 id LONG NOT NULL COMMENT,
 ......
)
USING DELTA
PARTITIONED BY (partition_column)
LOCATION 'path_to_your_delta'

Re: Tring to create incremental pipeline but fails when I try to use outputMode "update"

BorislavBlagoev — Thu, 20 Jan 2022 14:11:44 GMT

@Hubert Dudek It works like that. I have one more question. How can I include and delete that query?

  microBatchOutputDF._jdf.sparkSession().sql("""
    MERGE INTO old o
    USING updates u
    ON u.id= o.id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)

Or how I can add and delete rows from this pipeline.