dump delta gold table to cassandra table with delta only

Viewed 22

we have a Databricks workflow that run a delta live tables first then dump result from gold table to a cassandra table. when I ran the workflow i noticed it always dump all rows from gold table to cassandra table.

I understand when aggregate data from silver table and dump to gold table, it has to scan the whole silver table, but it doesn't mean every row changes in the gold table, how can I just dump the rows that were changed from gold to cassandra instead of dump all rows from gold to cassandra?

here are the code:

@dlt.table(comment="table with userId and payment method")
@dlt.expect_or_drop("supported event type", "event_type IN ('add','update')")
def test_kafka_silver():
  return (
    dlt.read_stream("test_kafka_bronze") \
       .withColumnRenamed("userId", "user_id") \
       .withColumnRenamed("type", "event_type") \
       .withColumnRenamed("actionDate", "event_at") \
       .select("user_id", "event_type", "event_at")
)

@dlt.table
def test_live_gold():
  return (
    dlt.read("test_kafka_silver").groupBy("user_id", "event_type").count()
  )

notebook to dump from gold to cassandra:

df = spark.read.format("delta").table("test_poc.test_live_gold")
     .withColumnRenamed("user_id", "account_id")
     .withColumnRenamed("event_type", "event_name")
write_to_cassandra_table('test', 'test_keyspace', df);
0 Answers
Related