Skip to content
TwinCommitDocs

Apache Spark

The TwinCommit Spark connector exposes a pipeline’s change feed as a Structured Streaming source. Offsets map directly to source log positions, so Spark’s own checkpointing gives your jobs exactly-once processing.

Supported: Apache Spark 3.5 and 4.x, Scala 2.13, PySpark.

Enable the change feed

Section titled “Enable the change feed”

On the TwinCommit side, use the spark target to serve the feed:

target:
  type: spark
  listen: ":9200"
  retention: 24h
  tls:
    cert_file: /etc/twincommit/tls.crt
    key_file: /etc/twincommit/tls.key

Add the connector

Section titled “Add the connector”
Terminal window
spark-submit --packages io.twincommit:twincommit-spark_2.13:1.4.0 job.py

PySpark

Section titled “PySpark”
changes = (
    spark.readStream.format("twincommit")
    .option("endpoint", "https://twincommit-host:9200")
    .option("token", dbutils.secrets.get("replication", "tc-token"))
    .option("tables", "SALES.ORDERS,SALES.ORDER_LINES")
    .load()
)


# columns: table, op, commit_ts, position, key (struct), after (struct), before (struct)
orders = changes.filter("table = 'SALES.ORDERS'").select("op", "after.*")


(orders.writeStream
    .format("delta")
    .option("checkpointLocation", "/chk/orders")
    .toTable("analytics.orders_changes"))

Scala

Section titled “Scala”
val changes = spark.readStream
  .format("twincommit")
  .option("endpoint", "https://twincommit-host:9200")
  .option("tables", "SALES.ORDERS")
  .load()


changes.writeStream
  .foreachBatch { (batch: DataFrame, id: Long) =>
    batch.createOrReplaceTempView("updates")
    batch.sparkSession.sql("""
      MERGE INTO analytics.orders t
      USING (SELECT after.* , op FROM updates WHERE table = 'SALES.ORDERS') s
      ON t.order_id = s.order_id
      WHEN MATCHED AND s.op = 'D' THEN DELETE
      WHEN MATCHED THEN UPDATE SET *
      WHEN NOT MATCHED AND s.op <> 'D' THEN INSERT *
    """)
  }
  .option("checkpointLocation", "/chk/orders-merge")
  .start()