Idempotente Data-Pipelines: Patterns für sichere Retries

# FALSCH — nicht-idempotenter Append
def load_daily_orders(date: str):
    df = extract_orders(date)
    df.to_sql("orders", engine, if_exists="append")  # Retry = Duplikate

# FALSCH — nicht-idempotente Aggregation
def aggregate_daily_sales(date: str):
    rows = db.execute(f"SELECT SUM(amount) FROM orders WHERE date = '{date}'").fetchone()
    db.execute(f"INSERT INTO daily_sales (date, revenue) VALUES ('{date}', {rows[0]})")
    # Retry = Duplikat-Zeilen in daily_sales

# FALSCH — side-effectful Writes ohne Deduplizierung
def send_events_to_kafka(date: str):
    events = db.execute(f"SELECT * FROM events WHERE date = '{date}'")
    for event in events:
        producer.send("events_topic", event)  # Retry = Duplikat-Messages

# PySpark — idempotenter Partition-Overwrite (Delta Lake / Spark)
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("idempotent_load").getOrCreate()

# Dynamic Partition-Overwrite — ersetzt nur geschriebene Partitionen
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

def load_orders_for_date(processing_date: str):
    # Idempotent: Re-Run für dasselbe Datum produziert dasselbe Resultat.
    df = spark.read.parquet(f"s3://raw/orders/date={processing_date}/")

    transformed = df.select(
        "order_id", "customer_id", "amount", "processing_date"
    )

    # Dynamic Overwrite: ersetzt nur die geschriebene Partition
    transformed.write \
        .format("delta") \
        .mode("overwrite") \
        .partitionBy("processing_date") \
        .save("s3://datalake/orders/")

    # Zweiter Lauf für dasselbe Datum -> selbes Resultat, keine Duplikate

# PySpark — idempotenter MERGE mit Delta Lake
from delta.tables import DeltaTable

def upsert_customers(new_data_df):
    # Idempotent: zweimal laufen mit demselben DataFrame produziert denselben Zielzustand.
    delta_table = DeltaTable.forPath(spark, "s3://datalake/customers/")

    (delta_table.alias("target")
        .merge(
            new_data_df.alias("source"),
            "target.customer_id = source.customer_id"
        )
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute()
    )
    # Zweiter Lauf mit denselben Daten: MATCH feuert, Target-Zeile wird mit
    # denselben Werten überschrieben. Net-Effekt: keine Änderung. Idempotent.

-- Spark SQL / Databricks SQL — MERGE-Statement
MERGE INTO customers AS target
USING new_customer_data AS source
  ON target.customer_id = source.customer_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;
-- Sicher mehrfach für dieselben Source-Daten ausführbar

# Python / SQLAlchemy — idempotentes Truncate-dann-Insert
from sqlalchemy import create_engine, text

def reload_product_catalog(products: list):
    # Idempotent: Truncate + Reload produziert denselben Zustand unabhängig vom Run-Count.
    # Nur für kleine Referenztabellen, wo Full-Reload akzeptabel ist.
    engine = create_engine("postgresql://user:pass@host/db")
    with engine.begin() as conn:  # begin() = automatic commit/rollback
        conn.execute(text("TRUNCATE TABLE product_catalog"))
        conn.execute(
            text("INSERT INTO product_catalog (id, name, category) VALUES (:id, :name, :category)"),
            products
        )
    # In Transaktion gewickelt: entweder beide TRUNCATE und INSERT gelingen, oder keines

# PySpark — Deduplizierung auf Natural-Key vor dem Write
from pyspark.sql.functions import row_number, desc
from pyspark.sql.window import Window

def deduplicate_and_load(raw_df, natural_keys: list, event_time_col: str):
    # Behält nur den neuesten Record pro Natural-Key.
    # Idempotent: gleiche Rohdaten -> gleicher deduplizierter Output.
    window = Window.partitionBy(*natural_keys).orderBy(desc(event_time_col))

    deduped = (
        raw_df
        .withColumn("_row_num", row_number().over(window))
        .filter("_row_num = 1")
        .drop("_row_num")
    )

    return deduped

# Usage
raw = spark.read.parquet("s3://raw/events/date=2024-03-15/")
clean = deduplicate_and_load(raw, natural_keys=["event_id"], event_time_col="ingested_at")
clean.write.format("delta").mode("overwrite").save("s3://datalake/events/")

# Python — Idempotency-Key für externen API-Write
import hashlib
import requests

def send_notification(user_id: str, message: str, event_date: str):
    # Idempotent: gleiches (user_id, event_date) sendet immer dieselbe Notification,
    # auch bei mehrfachem Aufruf.
    
    # Deterministischer Idempotency-Key aus stabilen Inputs
    idempotency_key = hashlib.sha256(
        f"{user_id}:{event_date}:{message}".encode()
    ).hexdigest()

    response = requests.post(
        "https://api.notifications.example.com/send",
        json={"user_id": user_id, "message": message},
        headers={
            "Idempotency-Key": idempotency_key,
            "Authorization": "Bearer TOKEN"
        }
    )

    if response.status_code == 200:
        return "sent"
    elif response.status_code == 409:
        return "already_sent"  # Duplikat erkannt — safe to ignore
    else:
        response.raise_for_status()

# PySpark Structured Streaming — exactly-once zu Delta Lake
query = (
    stream_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/orders_stream")
    # Delta Lake + Spark Structured Streaming = exactly-once by default
    # Checkpoint trackt Kafka-Offsets; Delta Lake nutzt Transaction-Log
    # für atomic Writes. Zusammen: exactly-once end-to-end.
    .start("s3://datalake/orders_stream/")
)

# pytest — Idempotenz-Test
def test_pipeline_is_idempotent(spark, test_data):
    # Pipeline zweimal laufen lassen produziert dasselbe Resultat wie ein Lauf.
    
    # Erster Lauf
    load_orders_for_date("2024-03-15")
    result_after_first_run = spark.read.format("delta").load("s3://test/orders/").count()

    # Zweiter Lauf — gleicher Input
    load_orders_for_date("2024-03-15")
    result_after_second_run = spark.read.format("delta").load("s3://test/orders/").count()

    assert result_after_first_run == result_after_second_run, \
        "Pipeline ist nicht idempotent: Row-Count änderte sich beim zweiten Lauf"