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

Spark Structured Streaming'i veritabanları veya API'ler gibi dış sistemlere veri yazmak için kullandığımda, iş yeniden başlatıldıktan sonra yinelenen kayıtlar görmeye devam ediyordum. Başta ağ veya sink konektöründen kaynaklandığını düşündüm. Ama checkpointing ve mikro-batch davranışını incelemeye başladığında, sorunun akışta değil, yazım şeklimde olduğunu fark ettim. İşte bu noktada foreachBatch, güvenilir ve güvenli yazılar için benim tercih ettiğim desen oldu.

Spark Structured Streaming'de foreachBatch, her mikro-batch'i standart batch işlemleriyle bir DataFrame veya Dataset olarak işlemenizi sağlar. Bu güçlü bir özellik çünkü yazım mantığını — işlemler, idempotency kontrolleri ve tam bir kez garantisi dahil — tam kontrol altında tutarken, akışın avantajlarından vazgeçmeden kullanmanızı mümkün kılar.

Pratikte nasıl uyguladığımı şöyle anlatırım. İlk olarak, bir batch DataFrame ve batch ID'sini alan bir fonksiyon tanımlarım:

void writeBatch(DataFrame batchDF, long batchId) {
  if (batchDF.isEmpty()) return;

  // batchId'yi idempotency için kullan — örneğin, işlenen batch ID'lerini bir deduplication tablosunda sakla
  val processed = spark.sql("SELECT batch_id FROM streaming_metadata WHERE batch_id = $batchId")
  if (processed.count() > 0) {
    log.info("Batch $batchId already processed. Skipping.");
    return;
  }

  try {
    batchDF.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "target_table")
      .option("user", username)
      .option("password", password)
      .mode(SaveMode.Append)
      .save()

    // Yazma başarılı olduktan sonra batch'i işlenmiş olarak işaretle
    spark.sql("INSERT INTO streaming_metadata (batch_id, processed_at) VALUES ($batchId, current_timestamp())")
  } catch {
    case e: Exception =>
      log.error("Failed to write batch $batchId", e)
      throw e  // Streaming'in checkpoint üzerinden yeniden denemesini sağla
  }
}

Ardından bu fonksiyonu akışa eklerim:

streamingDF.writeStream
  .foreachBatch(writeBatch _)
  .outputMode("Update")
  .option("checkpointLocation", "/mnt/checkpoints/streaming_write")
  .start()

Buradaki anahtar, Structured Streaming tarafından otomatik olarak sağlanan batchId'yi kullanmak — zaten işlenmiş mikro-batch'leri tespit etmek ve atlamak için. Bu, yazma işlemini idempotent hale getirir. Yaz-ahead logları ve checkpointing ile birleştiğinde, iş işinin mikro-batch'in ortasında başarısız olması durumunda bile tam bir kez garantisi sunar.

Birçok ekip, dış sistemlere yazma için foreach veya mapPartitions kullanmayı dener, ancak bu seviyede batch-level kontrol sağlamazlar. batchId olmadan, neyin işlendiğini güvenli bir şekilde takip edemezsiniz. Ben bunu zor yoldan öğrendim: bir Kafka konektörü arızası nedeniyle faturalandırma sistemimize 20 dakika boyunca yinelenen faturalar gelmeye başladı.

Birkaç pratik ipucu:

• İşlenen batch ID'lerini izlemek için kalıcı bir meta veri deposu (veritabanı tablosu veya Delta Lake) kullanın — bellek içi kümeler değil.
• Idempotency kontrolünü hafif tutun; batch'i yavaşlatan ağır sorgulardan kaçının.
• Dış sisteminiz işlem destekliyorsa (PostgreSQL, SQL Server gibi), yazma ve meta veri güncellemesini tek bir işlem içinde sarın.
• foreachBatch'te SaveMode.Overwrite'i, bir bölümü niyeteten değiştirmek istediğinizde dışında kullanmayın — risklidir ve akışın amacını yener.

Daha önce systemd dilimlerini izlemeyle ilgili yazımda bahsettiğim gibi, gözlemlenebilirlik kritiktir — bu yüzden her zaman batch işleme süresi, başarı/başarısızlık sayısı ve gecikme için metrikler eklerim. Bununla birlikte durmuş akışlar için uyarı kurarsanız, üretime hazır bir pipeline elde etmiş olursunuz.

REST API'leri veya eski veritabanları gibi tam bir kez desteği olmayan sistemlere yazıyorsanız, foreachBatch ve idempotency anahtarları genellikle en pratik çözümdür. Bu sihir değil — sadece akış çerçevesi içinde disiplinli batch işlemi. Ortamımda, bu yaklaşım verinin doğru olmasını ve uyarların sessiz kalmasını sağlar.

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