از snapshotهای روزانه تا جریان CDC (شامل INSERT / UPDATE / DELETE)

from pyspark.sql import functions as F

cles  = ["customer_id"]
suivi = [c for c in snap_today.columns if c not in cles]

def avec_hash(d):
    return d.withColumn("h", F.sha2(F.concat_ws("||", *suivi), 256))

auj, hier = avec_hash(snap_today).alias("n"), avec_hash(snap_yesterday).alias("o")

cdc = (
    auj.join(hier, cles, "full_outer")
       .withColumn("op",
           F.when(F.col("o.h").isNull(), "INSERT")
            .when(F.col("n.h").isNull(), "DELETE")
            .when(F.col("n.h") != F.col("o.h"), "UPDATE"))
       .filter("op IS NOT NULL")
)
cdc.groupBy("op").count().orderBy("op").show()