Spark Structured Streaming'de foreachBatch ile Güvenli Dış Yazılar

Ortamımda Spark Structured Streaming, gerçek zamanlı veri işlem hattları için tercih ettiğim araç. Ancak veritabanları veya API gibi dış sistemlere güvenli yazmak, bu hattın en zorlu kısmı hâlâ. Temel zorluk, hatalar oluştuğunda veri tutarlılığını sağlamak: yinelenen veri istemiyorum, kaybolan veri de olmamalı. İşte burada foreachBatch, doğru checkpointing ve idempotent yazılar devreye giriyor.

Why foreachBatch Over foreach

Structured Streaming ile ilk çalışmaya başladığımda, dış yazılar için foreach kullanmayı denedim. Basit geliyordu: her satır üzerinde döngü kur ve dış API'yi çağır. Ancak bu yaklaşım, yazma işlemi çevrimsel bir sınır içinde olmadığı için exact-once garantisini bozuyor. Bir görev yazdıktan sonra commit etmeden başarısız olursa, yeniden başlatmada yinelenen veri oluşur.

foreachBatch ise mikro-batch DataFrame'i ve batch ID'yi verir. Bu sayede net bir sınır elde edersin: tüm batch'i işle, sonra bir birim olarak commit veya roll back yap. Bu, idempotent mantığı uygulamayı kolaylaştırır — örneğin, dış sistemde zaten işlenmiş veriyi atlamak için benzersiz batch ID kullanmak gibi.

Implementing Idempotent Writes

Idempotent yazı, aynı veriyi birden fazla kez yazmanın tek kez yazmaya etkisi olması anlamına gelir. Uygulamada bunu Spark batch ID'sini yazma işlemine dahil ederek yaparım. Örneğin, PostgreSQL tablosuna yazarken batch_id sütunu ekliyorum ve bunu UPSERT'te kullanıyorum:

INSERT INTO events (event_id, payload, batch_id)
VALUES (?, ?, ?)
ON CONFLICT (event_id, batch_id) DO UPDATE SET payload = EXCLUDED.payload;

Bu sayede, bir hata nedeniyle aynı batch yeniden işlenirse, yazmanın netto etkisi yok olur. Zaman damgası veya satır düzeyi ID'leri tek başına kullanmaktan kaçınıyorum, çünkü bunlar yeniden denemeler arasında idempotency garantisi veremez.

Checkpointing for Recovery

Checkpointing müzakere edilemez olmaz. Olmazsa, Spark bir hata sonrası streaming sorgunuzun durumunu kurtaramaz. Her zaman güvenilir bir checkpoint konumu yapılandırırım — tercihen HDFS veya S3 gibi dağıtık bir dosya sistemi:

query = df.writeStream \
    .foreachBatch(write_to_external) \
    .outputMode("update") \
    .option("checkpointLocation", "s3://my-bucket/checkpoints/events-pipeline") \
    .start()

Checkpoint, sadece metadata değil, her mikro-batch'in ilerlemesini de saklar. Sürücü başarısız olursa, Spark son başarıyla commit edilmiş batch'ten devam edebilir — böylece veri kaybı veya yinelenmesi önlenir.

Handling Failures Gracefully

Idempotent yazılar ve checkpointing olsa da, geçici hatalarla başa çıkmak gerekir. Dış yazıyı network-level sorunlar için üst üste gebelikli geri dönüş ile bir retry döngüsüne sararım — veri doğrulama hataları için asla değil. Örneğin:

def write_to_external(batch_df, batch_id):
    try:
        # Convert to Pandas or use JDBC for batch write
        batch_df.write \
            .format("jdbc") \
            .option("url", jdbc_url) \
            .option("dbtable", "events") \
            .option("user", db_user) \
            .option("password", db_password) \
            .mode("append") \
            .save()
    except Exception as e:
        if "network" in str(e).lower() or "timeout" in str(e).lower():
            raise  # Spark'ın görev yeniden denemesini tetikle
        else:
            # Geçersiz veriyi logla ve batch'i başarısızlama — sadece boş batch onayla
            log_error(batch_id, e)
            batch_df.filter("false").write.format("noop").save()

Bu sayede geçici sorunlar Spark'ın yerleşik retry mekanizmasını tetikler, kötü veri ise tüm batch'i öldürmeden izole edilir.

Monitoring and Validation

Gecikme ve throughput'u Spark'ın akış metrikleriyle izlerim, ancak ayrıca uçtan uca doğrulama da eklerim. Kurulumumda her batch'in bir checksum'ını ayrı bir tabloda tutarım ve sessiz periyotlarda kaynakla karşılaştırırım. Eşleşmezse uyarı tetiklerim — bu, metrikler yakalayamayan sessiz hataları yakalamama yardımcı oldu.

Conclusion

foreachBatch ile idempotent yazılar ve doğru checkpointing kullanmak, sadece en iyi uygulama değil, Spark Structured Streaming'de güvenilir dış yazıların temeli. Dış işlem sistemlerine güvenmeden exact-once garantisi sağlar — bu, SaaS API'leri veya tutarlı olmayan depolama sistemleriyle çalışırken kritik. Argo Rollouts'teki latency SLO'larımdaki yazımda da belirttiğim gibi, veri katmanındaki tutarlılık doğrudan alt sistemlerin güvenilirliğini etkiler — bu yüzden bu kısmı doğru yapmak için zaman ayırmak değer.

Kapak görseli: learn_tek · CC0 (Openverse / kamu malı) · https://www.flickr.com/photos/153724200@N07/27774351928