upsert (merge) delta with spark structured streaming

Question

I need to upsert data in real time (with spark structured streaming) in python This data is read in realtime (format csv) and then is written as a delta table (here we want to update the data that's why we use merge into from delta) I am using delta engine with databricks I coded this:

from delta.tables import *
spark = SparkSession.builder \
  .config("spark.sql.streaming.schemaInference", "true")\
  .appName("SparkTest") \
  .getOrCreate()
sourcedf= spark.readStream.format("csv") \
  .option("header", True) \
  .load("/mnt/user/raw/test_input") #csv data that we read in real time
spark.conf.set("spark.sql.shuffle.partitions", "1")

spark.createDataFrame([], sourcedf.schema) \
  .write.format("delta") \
  .mode("overwrite") \
  .saveAsTable("deltaTable")

def upsertToDelta(microBatchOutputDF, batchId): 
  microBatchOutputDF.createOrReplaceTempView("updates")
  microBatchOutputDF._jdf.sparkSession().sql("""
    MERGE INTO deltaTable t
    USING updates s
    ON s.Id = t.Id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)

sourcedf.writeStream \
  .format("delta") \
  .foreachBatch(upsertToDelta) \
  .outputMode("update") \
  .option("checkpointLocation", "/mnt/user/raw/checkpoints/output")\
  .option("path", "/mnt/user/raw/PARQUET/output") \
  .start() \ 
  .awaitTermination()

but nothing gets written as expected in the output path , the checkpoint path gets filled in as expected , a display in the delta table gives me results too

display(table("deltaTable"))

in the spark UI I see the writestream step :

sourcedf.writeStream \ .format("delta") \ ....
first at Snapshot.scala:156+details
RDD: Delta Table State #1 - dbfs:/user/hive/warehouse/deltatable/_delta_log

any idea how to fix this so I can upsert csv data into delta tables in S3 in real time with spark Best regards

user_19550042 · Answer

Apologies for a late reply, but just in case anyone else has the same problem. I have found the below worked for me, I wonder is it because you didn't use "cloudFiles" on your readstream to make use of autoloader?:

%python
sourcedf= spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("cloudFiles.includeExistingFiles","true") \
  .schema(csvSchema) \
  .load("/mnt/user/raw/test_input")

%sql
CREATE TABLE IF NOT EXISTS deltaTable(
  col1 int NOT NULL,
  col2 string NOT NULL,
  col3 bigint,
  col4 int
  )
  USING DELTA
  LOCATION '/mnt/user/raw/PARQUET/output'

%python
def upsertToDelta(microBatchOutputDF, batchId): 
  microBatchOutputDF.createOrReplaceTempView("updates")
  microBatchOutputDF._jdf.sparkSession().sql("""
    MERGE INTO deltaTable t
    USING updates s
    ON s.Id = t.Id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
  """)

%python
sourcedf.writeStream \
  .format("delta") \
  .foreachBatch(upsertToDelta) \
  .outputMode("update") \
  .option("checkpointLocation", "/mnt/user/raw/checkpoints/output") \
  .start("/mnt/user/raw/PARQUET/output")

upsert (merge) delta with spark structured streaming

1 Answers