Delta Lake Event Driven Design — ออกแบบ Data

Delta Lake Event Driven

Delta Lake Event Driven ACID Transaction Change Data Feed Streaming Spark Merge Upsert Schema Evolution Time Travel S3 ADLS GCS

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน Immutable OS Fedora CoreOS Cloud Migration

Feature	Delta Lake	Apache Iceberg	Apache Hudi
ACID Transactions	Yes	Yes	Yes
Change Data Feed	CDF (Built-in)	Incremental Read	CDC (Built-in)
Streaming	Spark Structured Streaming	Flink	Spark/Flink
Merge/Upsert	MERGE INTO	MERGE INTO	Upsert
Time Travel	Version/Timestamp	Snapshot	Timeline
Optimize	OPTIMIZE + Z-ORDER	Compaction	Compaction

Change Data Feed & Merge

# === Change Data Feed & Merge Operations ===



# Read Change Data Feed

# changes_df = spark.read \

#     .format("delta") \

#     .option("readChangeFeed", "true") \

#     .option("startingVersion", 5) \

#     .option("endingVersion", 10) \

#     .table("events")

#

# # CDF columns: _change_type, _commit_version, _commit_timestamp

# # _change_type: insert, update_preimage, update_postimage, delete

#

# inserts = changes_df.filter("_change_type = 'insert'")

# updates = changes_df.filter("_change_type = 'update_postimage'")

# deletes = changes_df.filter("_change_type = 'delete'")



# Merge (Upsert) - Exactly-once Processing

# deltaTable = DeltaTable.forPath(spark, "delta://silver/users")

# deltaTable.alias("target").merge(

#     source_df.alias("source"),

#     "target.user_id = source.user_id"

# ).whenMatchedUpdate(set={

#     "name": "source.name",

#     "email": "source.email",

#     "updated_at": "source.event_time"

# }).whenNotMatchedInsert(values={

#     "user_id": "source.user_id",

#     "name": "source.name",

#     "email": "source.email",

#     "created_at": "source.event_time",

#     "updated_at": "source.event_time"

# }).execute()



@dataclass

class MergePattern:

    pattern: str

    use_case: str

    match_condition: str

    matched_action: str

    not_matched_action: str



patterns = [

    MergePattern("SCD Type 1 (Overwrite)",

        "อัพเดทข้อมูลล่าสุด ไม่เก็บประวัติ",

        "target.id = source.id",

        "UPDATE SET * (Overwrite ทุก Column)",

        "INSERT * (Insert Row ใหม่)"),

    MergePattern("SCD Type 2 (History)",

        "เก็บประวัติทุก Version",

        "target.id = source.id AND target.is_current = true",

        "UPDATE SET is_current=false, end_date=now()",

        "INSERT with is_current=true, start_date=now()"),

    MergePattern("Deduplication",

        "ลบ Duplicate Event",

        "target.event_id = source.event_id",

        "ไม่ทำอะไร (Skip)",

        "INSERT * (Insert เฉพาะ Event ใหม่)"),

    MergePattern("Delete + Insert",

        "Replace Partition ด้วยข้อมูลใหม่",

        "target.date = source.date",

        "DELETE (ลบ Partition เก่า)",

        "INSERT * (Insert ข้อมูลใหม่ทั้ง Partition)"),

]



print("=== Merge Patterns ===")

for m in patterns:

    print(f"\n  [{m.pattern}] {m.use_case}")

    print(f"    Match: {m.match_condition}")

    print(f"    Matched: {m.matched_action}")

    print(f"    Not Matched: {m.not_matched_action}")