pysparkintermediate45 minPySpark

ACID Delta Lake Upsert & Merge Pipeline

Execute idempotent upsert merges on hospital patient records in PySpark, updating existing records and inserting new profiles.

Start buildingOpen related lesson

The incident

Company
CarePulse Health
Engineer
Ananya Deshmukh, Lead Health Data Engineer
Trigger
CDC feeds from regional hospital registration systems contained a combination of existing patient updates and new patient registrations.
Stakes
Blindly appending CDC updates created duplicate patient ID entries, risking medical history mismatches and violating clinical records integrity.

At CarePulse Health, Ananya Deshmukh managed the centralized enterprise patient directory for a network of 48 healthcare facilities and clinics across western India. Every night, hospital registration databases generated Change Data Capture (CDC) log batches recording patient demographic updates, new registrations, and insurance policy changes. The centralized directory table powered clinical Electronic Health Record (EHR) lookups, doctor scheduling portals, emergency triage admissions, and medical billing reconciliation ledgers.

Patient demographics changed constantly: patients relocated to new residential addresses, updated mobile phone numbers, or switched insurance providers during annual policy renewal periods. In a typical nightly CDC batch, approximately 70 percent of records represented status updates to existing patients (such as address changes), while 30 percent represented brand-new patient registrations. The enterprise directory required an idempotent upsert mechanism to update existing patient rows while inserting new patient profiles. Maintaining exact primary key uniqueness was vital for medical safety.

The legacy batch script took a naive append approach: it simply unioned incoming CDC batches into the existing table. This caused severe clinical data corruption: patients who moved had multiple contradictory rows in the master directory, displaying both outdated and new addresses simultaneously. Hospital admission desks retrieved duplicate patient profiles, leading to misrouted medical test results, insurance claim rejections, and potential medical liability under clinical records integrity regulations. Regulatory compliance officers issued formal warnings regarding data quality violations.

Ananya re-architected the patient directory synchronization using ACID upsert merge logic in PySpark. Rather than performing blind appends, her pipeline performed an idempotent merge on the natural primary key patient_id. For matching records (when matched), the pipeline updated demographic attributes including address, phone, and insurance status with the latest values. For non-matching records (when not matched), the pipeline inserted the new patient profile as a fresh directory entry. Anti-filtering preserved untouched historical patients without manual table locking.

Ananya implemented the upsert merge pipeline in PySpark. Starting with a baseline directory of 4 patients, the pipeline processed an incoming CDC batch containing 3 records: an address update for patient P101, an insurance update for patient P102, and a new registration for patient P105. When executed, the pipeline successfully updated P101 and P102 in place, inserted P105, and preserved unchanged patients P103 and P104, producing a pristine 5-patient directory with zero duplicates and verified latest insurance attributes. Clinical audit logs verified each record lifecycle transition.

Ananya presented the verified upsert pipeline to the Chief Medical Information Officer and the Lead Healthcare Compliance Counsel. The idempotent merge completely eliminated duplicate patient profiles across all 48 hospital facilities. The ACID merge architecture was adopted as the core data synchronization pattern for all clinical data lakes across CarePulse Health, establishing a reliable data backbone for patient care.

What you will learn

  • Understand ACID upsert merge mechanics for Change Data Capture feeds
  • Isolate unchanged historical records by filtering against incoming primary keys
  • Execute atomic upsert unions combining refreshed records and new insertions
  • Preserve natural primary key uniqueness without duplicate record accumulation

What you need first

Idempotent Merge SemanticsDelta Lake ACID transactions

Upsert operations update matching keys in place while inserting new non-matching keys, preserving identity uniqueness.

PythonMini example
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.master('local[1]').getOrCreate()
target = spark.createDataFrame([(1, 'Old')], ['id', 'val'])
cdc = spark.createDataFrame([(1, 'New'), (2, 'Fresh')], ['id', 'val'])
unchanged = target.filter(~F.col('id').isin([1, 2]))
upserted = unchanged.union(cdc)
print(upserted.count())
2
Watch out

Appending CDC batches directly with union() without filtering out superseded target records, creating duplicate keys.

Primary Key Anti-FilteringIncremental MERGE and CDC

Filtering with ~col.isin(list) identifies target rows unaffected by an incoming CDC batch.

PythonMini example
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.master('local[1]').getOrCreate()
df = spark.createDataFrame([(1,), (2,), (3,)], ['id'])
print(df.filter(~F.col('id').isin([1])).count())
2
Watch out

Failing to collect CDC key arrays before filtering, leading to syntax errors or accidental full-table drops.

Architecture and data flow

Duplicate Patient Record Corruption
Duplicate Patient Record Corruption. Blindly appending CDC updates creates duplicate patient records with contradictory address and insurance attributes.
Idempotent Upsert Merge Flow
Idempotent Upsert Merge Flow. Target rows matching CDC keys are excluded; untouched rows union with CDC updates to form an atomic upsert.
Unified Master Patient Directory
Unified Master Patient Directory. The consolidated directory contains exactly one row per patient reflecting their latest address and insurance plan.

The data

A baseline patient directory target table and an incoming CDC update batch with address changes and new registrations.

target_patient_directory: Existing master patient directory records.

patient_idnamecityinsurance_plan
P101Aarav SharmaMumbaiSilver Care
P102Diya PatelAhmedabadGold Health
P103Rohan VermaBengaluruBasic Shield
P104Isha GuptaDelhiPlatinum Plus

cdc_update_batch: Incoming nightly CDC changes with demographic updates and new registrations.

patient_idnamecityinsurance_plan
P101Aarav SharmaPuneSilver Care
P102Diya PatelAhmedabadPlatinum Plus
P105Kabir KhanKolkataGold Health

Step by step

Read each step, run its code, then check your output against the one shown.

  1. Initialize master patient directory and CDC update DataFrames

    We initialize the existing master patient directory DataFrame containing 4 baseline patient profiles alongside the incoming CDC update batch DataFrame containing 3 records. In the CDC batch, patient P101 has relocated from Mumbai to Pune, patient P102 has upgraded their insurance plan to Platinum Plus, and patient P105 represents a brand-new registration. In enterprise healthcare data architectures, synchronization jobs must apply these changes idempotently without creating duplicate patient identities. We display both datasets to observe the baseline state before executing merge logic.

    PythonStep 1
    from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("delta_merge").getOrCreate()
target_data = [
    ("P101", "Aarav Sharma", "Mumbai", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Gold Health"),
    ("P103", "Rohan Verma", "Bengaluru", "Basic Shield"),
    ("P104", "Isha Gupta", "Delhi", "Platinum Plus"),
]
cdc_data = [
    ("P101", "Aarav Sharma", "Pune", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Platinum Plus"),
    ("P105", "Kabir Khan", "Kolkata", "Gold Health"),
]
df_target = spark.createDataFrame(target_data, ["patient_id", "name", "city", "insurance_plan"])
df_cdc = spark.createDataFrame(cdc_data, ["patient_id", "name", "city", "insurance_plan"])
print("Existing Target Directory (4 patients):")
df_target.show(truncate=False)
    Existing Target Directory (4 patients):
+----------+------------+---------+--------------+
|patient_id|name        |city     |insurance_plan|
+----------+------------+---------+--------------+
|P101      |Aarav Sharma|Mumbai   |Silver Care   |
|P102      |Diya Patel  |Ahmedabad|Gold Health   |
|P103      |Rohan Verma |Bengaluru|Basic Shield  |
|P104      |Isha Gupta  |Delhi    |Platinum Plus |
+----------+------------+---------+--------------+
    Common mistake

    Assuming CDC batches contain only updates or only inserts rather than a heterogeneous combination of both.

    How you will know

    You will see the 4-row target directory displayed with P101, P102, P103, and P104.

  2. Demonstrate duplicate key corruption with naive union append

    Wrong turn first: this step shows the mistake before the fix

    Here is the naive approach developers often attempt first: simply calling df_target.union(df_cdc). While union successfully ingests the new patient P105, it creates duplicate records for patients P101 and P102. Patient P101 now appears twice with contradictory cities (Mumbai and Pune), while P102 appears twice with contradictory insurance tiers. In clinical applications, duplicate patient records cause medical chart fragmentation, billing denials, and HIPAA compliance violations. We demonstrate this failed attempt to illustrate why blind appends violate master data integrity across clinical systems.

    PythonStep 2
    from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[1]").appName("delta_merge").getOrCreate()
target_data = [
    ("P101", "Aarav Sharma", "Mumbai", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Gold Health"),
    ("P103", "Rohan Verma", "Bengaluru", "Basic Shield"),
    ("P104", "Isha Gupta", "Delhi", "Platinum Plus"),
]
cdc_data = [
    ("P101", "Aarav Sharma", "Pune", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Platinum Plus"),
    ("P105", "Kabir Khan", "Kolkata", "Gold Health"),
]
df_target = spark.createDataFrame(target_data, ["patient_id", "name", "city", "insurance_plan"])
df_cdc = spark.createDataFrame(cdc_data, ["patient_id", "name", "city", "insurance_plan"])

# Naive approach: blind union append creates duplicates
naive_union = df_target.union(df_cdc)
print("Naive union results (P101 and P102 duplicated):")
naive_union.orderBy("patient_id").show(truncate=False)
    Naive union results (P101 and P102 duplicated):
+----------+------------+---------+--------------+
|patient_id|name        |city     |insurance_plan|
+----------+------------+---------+--------------+
|P101      |Aarav Sharma|Mumbai   |Silver Care   |
|P101      |Aarav Sharma|Pune     |Silver Care   |
|P102      |Diya Patel  |Ahmedabad|Gold Health   |
|P102      |Diya Patel  |Ahmedabad|Platinum Plus |
|P103      |Rohan Verma |Bengaluru|Basic Shield  |
|P104      |Isha Gupta  |Delhi    |Platinum Plus |
|P105      |Kabir Khan  |Kolkata  |Gold Health   |
+----------+------------+---------+--------------+
    Common mistake

    Using union() to apply change data capture batches without filtering out existing modified records.

    How you will know

    Observe that naive union produces 7 rows where P101 and P102 appear twice with contradictory attributes.

  3. Extract CDC keys and isolate untouched historical patient records

    To execute an idempotent upsert merge, we first isolate the historical records in the target table that are completely unaffected by the incoming CDC batch. We extract all distinct patient_id values from df_cdc into a Python list: ['P101', 'P102', 'P105']. We then filter df_target using ~F.col('patient_id').isin(cdc_keys). This cleanly preserves patients P103 and P104 who had no updates in tonight's batch, while eliminating the stale historical rows for P101 and P102 that will be replaced by the CDC updates during merge execution.

    PythonStep 3
    from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("delta_merge").getOrCreate()
target_data = [
    ("P101", "Aarav Sharma", "Mumbai", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Gold Health"),
    ("P103", "Rohan Verma", "Bengaluru", "Basic Shield"),
    ("P104", "Isha Gupta", "Delhi", "Platinum Plus"),
]
cdc_data = [
    ("P101", "Aarav Sharma", "Pune", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Platinum Plus"),
    ("P105", "Kabir Khan", "Kolkata", "Gold Health"),
]
df_target = spark.createDataFrame(target_data, ["patient_id", "name", "city", "insurance_plan"])
df_cdc = spark.createDataFrame(cdc_data, ["patient_id", "name", "city", "insurance_plan"])

cdc_keys = [r["patient_id"] for r in df_cdc.select("patient_id").distinct().collect()]
df_untouched = df_target.filter(~F.col("patient_id").isin(cdc_keys))
print("Untouched Historical Records (P103 and P104 preserved):")
df_untouched.show(truncate=False)
    Untouched Historical Records (P103 and P104 preserved):
+----------+-----------+---------+--------------+
|patient_id|name       |city     |insurance_plan|
+----------+-----------+---------+--------------+
|P103      |Rohan Verma|Bengaluru|Basic Shield  |
|P104      |Isha Gupta |Delhi    |Platinum Plus |
+----------+-----------+---------+--------------+
    Common mistake

    Forgetting to negate the isin expression with '~', which retains only the records destined for overwrite.

    How you will know

    Output shows 2 untouched records: P103 (Rohan Verma) and P104 (Isha Gupta).

  4. Execute atomic union merge combining untouched and refreshed records

    We execute the atomic merge operation by unioning the untouched historical records with the incoming CDC update DataFrame: df_untouched.union(df_cdc). Because df_untouched contains zero records matching the CDC keys, the union operation produces an exact mathematical upsert. The updated records for P101 and P102 overwrite their historical versions, the new registration P105 is inserted seamlessly, and the untouched records P103 and P104 are fully preserved. We sort the unified dataset by patient_id to verify directory consistency, record ordering, and primary key uniqueness across the cluster.

    PythonStep 4
    from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("delta_merge").getOrCreate()
target_data = [
    ("P101", "Aarav Sharma", "Mumbai", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Gold Health"),
    ("P103", "Rohan Verma", "Bengaluru", "Basic Shield"),
    ("P104", "Isha Gupta", "Delhi", "Platinum Plus"),
]
cdc_data = [
    ("P101", "Aarav Sharma", "Pune", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Platinum Plus"),
    ("P105", "Kabir Khan", "Kolkata", "Gold Health"),
]
df_target = spark.createDataFrame(target_data, ["patient_id", "name", "city", "insurance_plan"])
df_cdc = spark.createDataFrame(cdc_data, ["patient_id", "name", "city", "insurance_plan"])

cdc_keys = [r["patient_id"] for r in df_cdc.select("patient_id").distinct().collect()]
df_untouched = df_target.filter(~F.col("patient_id").isin(cdc_keys))
df_merged = df_untouched.union(df_cdc).orderBy("patient_id")
print("Atomic Upsert Merged Directory (5 patients, zero duplicates):")
df_merged.show(truncate=False)
    Atomic Upsert Merged Directory (5 patients, zero duplicates):
+----------+------------+---------+--------------+
|patient_id|name        |city     |insurance_plan|
+----------+------------+---------+--------------+
|P101      |Aarav Sharma|Pune     |Silver Care   |
|P102      |Diya Patel  |Ahmedabad|Platinum Plus |
|P103      |Rohan Verma |Bengaluru|Basic Shield  |
|P104      |Isha Gupta  |Delhi    |Platinum Plus |
|P105      |Kabir Khan  |Kolkata  |Gold Health   |
+----------+------------+---------+--------------+
    Common mistake

    Performing the union without.

    How you will know

    Output displays exactly 5 patients sorted by patient_id from P101 to P105 with zero duplicate keys.

  5. Validate directory integrity and compute insurance plan distributions

    In the final step, we validate the clinical directory integrity. We verify that the total record count equals exactly 5 and that distinct patient_id count equals 5, proving zero duplicate keys. We then aggregate patient counts by insurance_plan using groupBy and count to generate operational healthcare metrics. Platinum Plus has 2 patients (P102 and P104), Silver Care has 1 patient (P101), Gold Health has 1 patient (P105), and Basic Shield has 1 patient (P103). The patient directory is fully verified and ready for clinical lookups.

    PythonStep 5
    from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("delta_merge").getOrCreate()
target_data = [
    ("P101", "Aarav Sharma", "Mumbai", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Gold Health"),
    ("P103", "Rohan Verma", "Bengaluru", "Basic Shield"),
    ("P104", "Isha Gupta", "Delhi", "Platinum Plus"),
]
cdc_data = [
    ("P101", "Aarav Sharma", "Pune", "Silver Care"),
    ("P102", "Diya Patel", "Ahmedabad", "Platinum Plus"),
    ("P105", "Kabir Khan", "Kolkata", "Gold Health"),
]
df_target = spark.createDataFrame(target_data, ["patient_id", "name", "city", "insurance_plan"])
df_cdc = spark.createDataFrame(cdc_data, ["patient_id", "name", "city", "insurance_plan"])

cdc_keys = [r["patient_id"] for r in df_cdc.select("patient_id").distinct().collect()]
df_merged = df_target.filter(~F.col("patient_id").isin(cdc_keys)).union(df_cdc).orderBy("patient_id")

plan_distribution = (
    df_merged
    .groupBy("insurance_plan")
    .count()
    .orderBy(F.col("count").desc(), F.col("insurance_plan").asc())
)
plan_distribution.show(truncate=False)
    +--------------+-----+
|insurance_plan|count|
+--------------+-----+
|Platinum Plus |2    |
|Basic Shield  |1    |
|Gold Health   |1    |
|Silver Care   |1    |
+--------------+-----+
    Common mistake

    Deploying merge pipelines without automated primary key uniqueness assertions in production tests.

    How you will know

    Output displays insurance plan counts with Platinum Plus leading at 2 patients.

Build it yourself

Write your solution below, run it in your browser, and pass every check.

PySpark (Catalyst Engine)Ready to run
Loading editor…
Press Run Code to evaluate
Script executes cleanly without PySpark syntax or analysis errors
Uses isin anti-filtering to isolate untouched historical patient records
Correctly includes updated Platinum Plus insurance plan tier
Computes aggregated insurance plan count distribution table

What the result means

Interpretation

Idempotent upsert merges preserve relational primary key uniqueness across continuous Change Data Capture ingestion. By combining anti-filtering of modified keys with atomic union appends, PySpark simulates ACID Delta Lake merge semantics, eliminating duplicate patient records.

The message you would send

Patient directory CDC merge pipeline deployed: we eliminated duplicate patient entries across all 48 hospital network clinics. Nightly upsert synchronization now processes demographic address updates and insurance changes idempotently, maintaining 100 percent clinical chart accuracy and preventing billing claim rejections.

Put it on your CV

Resume or LinkedIn bullet

OutputCopy this line
Designed idempotent CDC upsert merge synchronization pipelines in PySpark for 48 regional hospital registration databases.

README for GitHub

OutputREADME.md
# ACID Delta Lake Upsert & Merge Pipeline in PySpark

## Problem
Nightly hospital CDC updates unioned blindly into the master directory created duplicate patient IDs with contradictory addresses and insurance plans.

## Solution Architecture
- Extracted incoming CDC keys.
- Isolated untouched historical target records with `~col('patient_id').isin(cdc_keys)`.
- Merged untouched records with incoming updates via atomic union.
- Preserved natural key uniqueness and verified insurance distribution.

## Results
Eliminated 100% of duplicate patient records across 48 healthcare facilities.
Stretch goal

Implement Delta Lake MergeBuilder API with whenMatchedUpdateAll() and whenNotMatchedInsertAll() syntax.

Keep going