actualizar (fusionar) delta con streaming estructurado de spark

Necesito insertar/actualizar datos en tiempo real (con transmisión estructurada de Spark) en Python.
Estos datos se leen en tiempo real (formato csv) y luego se escriben como una tabla delta (aquí queremos actualizar los datos, por eso usamos merge into desde delta).
Estoy usando el motor delta con Databricks.
He codificado lo siguiente:

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 que leemos en tiempo real
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()

Pero nada se escribe como se esperaba en la ruta de salida, la ruta del punto de control se llena como se esperaba, y una visualización de la tabla delta me muestra resultados también.

display(table("deltaTable"))

En la interfaz de usuario de Spark, veo el paso de escritura de transmisión:

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

¿Alguna idea de cómo solucionar esto para poder insertar/actualizar datos csv en tablas delta en S3 en tiempo real con Spark?
Saludos cordiales