How can I dedupe from a table created from a Kinesis change data capture feed.

[aliacovella]

Here I have a table named organizations_silver that was build from a bronze table created from a Kinesis change data capture feed.

@dlt.table(name="kinesis_raw_stream", table_properties={"pipelines.reset.allowed": "false"})
def kinesis_raw_stream():
    return read_kinesis_stream(stream_name=STREAM, dbutils=dbutils, spark=spark)

The initial table is an append table from the Kinesis stream. From there I destructured the data into the organizations_silver table. 

@dlt.table(name="organizations_silver")
def organizations_silver():
    org_schema = StructType([
        StructField("id", IntegerType(), True),
        StructField("org_name", StringType(), True),
        StructField("region", StringType(), True)
    ])

    return (
        dlt.readStream("kinesis_raw_stream")
        .select(decode(col("data"), "UTF-8").alias("json_string"))
        .select(from_json(col("json_string"), raw_schema).alias("json_data"))
        .filter(col("json_data.metadata.table-name") == "organizations")
        .select(from_json(col("json_data.data"), org_schema).alias("org_data"))
        .select("org_data.id", "org_data.org_name", "org_data.region")
    )

The problem is that since the original table is an append-only, records get duplicated in the silver table. 

I attempted to create another table from the silver table to dedupe it as follows:

df = spark.table("development_demo_catalog.default.organizations_silver")

df.write.format("delta").mode("overwrite").saveAsTable("development_demo_catalog.default.organizations_deduped")

DeltaTable.forName(spark, "development_demo_catalog.default.organizations_deduped") \
    .alias("target") \
    .merge(
        df.alias("source"),
        "source.id = target.id"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

My understanding is that if there is a match it should update all the columns on the existing deduped table. However, when I run the notebook, I encounter the following error:

Cannot perform Merge as multiple source rows matched and attempted to modify the same

This seems to indicate that there are duplicates in the source, which is the case and is why I'm trying to dedupe it. My question is what am I missing here and how should I properly dedupe records from a Kinesis change data capture feed?

[Alberto_Umana]

Hello @aliacovella,

Looks like there are duplicate records in your source table that match the same target record. This is indeed the case since your source table, organizations_silver, contains duplicates due to the append-only nature of the Kinesis stream.

To properly deduplicate records from a Kinesis change data capture feed, you can use the APPLY CHANGES API provided by Delta Live Tables (DLT). This API simplifies change data capture (CDC) by handling out-of-sequence records and ensuring correct processing of CDC records.

https://docs.databricks.com/en/delta-live-tables/cdc.html