Ortamımda gerçek zamanlı veri hatları, ağ gecikmeleri, cihaz tamponlaması veya saat dilimi farklılıkları nedeniyle gelen olaylar geç kalıyor. Spark Structured Streaming kullanıyorsan ve bunu görmezden geliyorsan, verileri sessizce kaybetme ya da yanlış agregasyonlar üretme riskini taşıyorsun. E-ticaret hatlarında mobil uygulamalardan gelen sipariş olaylarının dakikalarca geciktiğini ve gelir raporlarının eksik çıktığını gördüm. Düzelmesi karmaşık değil ama iki temel kavramı anlamak gerekiyor: watermark ve allowedLateness.
What is a watermark in Spark Structured Streaming?
Bir watermark, Spark'ın geç veriyi işlemek için ne kadar bekleyeceğini ve bu sürenin ardından veriyi çok geç görüp atacağını tanımlar. Bir olay-zaman sütununa watermark koyduğunda, Spark şu ana kadar gördüğü maksimum olay zamanını takip eder ve gecikme eşiğini çıkarır. Güncel watermark'tan daha eski olay zamanına sahip herhangi veri atılır. Bu, sonsuz durum büyümesini önler ve Spark'ın sonuçları finalize etmesine olanak tanır.
Örneğin, olayların zaman damgasından 2 dakika içinde gelmesi gerekiyorsa, watermark'ı "2 dakika" olarak ayarlarsın. Spark, olay zamanına göre son 2 dakika içindeki olaylar için sadece durum tutar ve eski durumu güvenli bir şekilde temizler.
How allowedLateness complements watermark
Watermark Spark'a geç veriyi ne zaman atacağını söylerken, allowedLateness, watermark'ın ötesinde sonuçları güncellemek için toler edilecek geç kalma miktarını belirler. Spark'ın durum tutma süresini uzatmaz — bu hâlâ watermark tarafından yönetilir — ancak allowedLateness penceresi içinde gelen geç verilerin zaten işlenmiş sonuçları güncellemesine olanak tanır.
Bunu şöyle düşünebilirsin: watermark "Bu noktadan sonra durum takibini durdurur" derken, allowedLateness "ama bu noktadan hemen sonra gelen veriler çıktıları güncellemek için kullanılsın" der. Bu, oturumlaştırma veya gerçek zamanlı kontrol panelleri gibi kullanım durumları için kritiktir; çünkü geciken olaylar geldiğinde son sonuçları düzeltmek istersin.
Practical example: processing orders with 3-minute latency tolerance
Geç gelen sipariş olaylarını işleyen bir akış sorgusunu şöyle yapılandırıyorum:
val orderEvents = spark
.readStream
.format("kafka")
.option("subscribe", "orders")
.option("startingOffsets", "latest")
.load()
.selectExpr("CAST(value AS STRING)")
.select(from_json(col("value"), orderSchema).as("data"))
.select("data.order_id", "data.amount", "data.event_time")
.withWatermark("event_time", "3 minutes")
val hourlySales = orderEvents
.groupBy(
window(col("event_time"), "1 hour"),
col("order_id")
)
.agg(sum("amount").as("total_amount"))
val query = hourlySales
.writeStream
.outputMode("update")
.format("console")
.option("truncate", "false")
.trigger(Trigger.ProcessingTime("1 minute"))
.option("allowedLateness", "3 minutes")
.start()Bu kurulumda:
- "3 dakika" watermark'ı, Spark'ın (maksimum olay zamanı - 3 dakika) değerinden daha eski event_time'ye sahip order_event'leri attığı anlamına gelir.
- "3 dakika" allowedLateness değeri, watermark'ın ötesinde en fazla 3 dakika geç gelen olayların pencerelik agregasyonları güncellemesine izin verir.
allowedLateness olmadan, geç veri watermark geçtikten sonra göz ardı edilir. Watermark olmadan, Spark tüm geçmiş veriyi tutmaya çalıştığı için durum sınırsız büyür.
Common pitfalls I’ve encountered
Sık gördüğüm bir hata, allowedLateness'i watermark'tan uzun ayarlamaktır. Bu anlamlı değil — allowedLateness, watermark'tan aşamaz çünkü Spark zaten o durumu attı. Watermark'ı "1 dakika" ve allowedLateness'i "5 dakika" olarak ayarlarsan, Spark 1 dakika gecikmeden sonra veriyi hâlâ atar; ekstra 4 dakika etkisiz olur.
Başka bir sorun, işleme zamanını olay zamanuyla karıştırmaktır. Zaman damgalarınız eğikse (örneğin cihazların saatleri yanlışsa), watermark mantığınız bozulur. Her zaman event_time alımını doğrula — ben işleme zamanı ile event_time arasındaki gecikmeyi ölçmek için basit bir metrik kullandım ve saat sapmasını bu şekilde tespit ettim.
Monitoring and tuning
Spark UI ve loglarındaki şunları izliyorum:
- StreamingQueryListener'dan
maxEventTimevewatermarkmetrikleri - Watermark nedeniyle atılan satır sayısı (son Spark sürümlerinde mevcut)
- Bölümlerdeki en son event_time ile geçerli zaman arasındaki gecikme
Düşülen satır sayısında artık bir artış görürsen, watermark'ın çok sıkı olabilir. Durum boyutu sürekli büyüyorsa, watermark çok gevşek veya eksik olabilir. Gözlemlediğin gecikme dağılımına göre 30 saniyelik artışlarla ayarlama yap.
When to use this pattern
Bu yaklaşım en iyi şu durumlarda çalışır:
- Gecikme için üst sınır biliniyorsa (örneğin "%99 olay 2 dakika içinde geliyor")
- Sonuçlara artımlı güncelleme ihtiyacı varsa (outputMode = "update")
- Son pencereler için küçük düzeltmelere tolerans gösteriliyorsa
Finansal mutabakat gibi gecikme mutlak sıfır olmalı senaryolarda, geç veriyi toplu onarım için dead-letter kuyruğuna yönlendirebilirsin. Ancak çoğu operasyonel kontrol paneli ve KPI için, watermark + allowedLateness doğruluk ve verimlilik arasında doğru dengeyi sağlar.
Final thoughts
Kafka, Kinesis ve Delta Lake kaynaklarıyla üretim ortamlarında bu deseni kullandım. Sadece veri kaybını önlemekle kalmaz, gerçek zamanlı sistemlere güveni artırır. Paydaşlar geciken verilere uygun olarak sayıların düzelttiğini gördüğünde, hattın güvenilirliğini sorgulamayı bırakırlar.
Structured Streaming ile başlıyorsan, önce latency SLA'nı belirle. Sonra buna uygun watermark ayarla ve çıktı düzeltmeleri için küçük bir tampon olmak üzere allowedLateness kullan. Yük altında davranışı doğrulamak için gecikme enjektörü aracıyla test et (Kafka'nın kafka-producer-perf-test ile özel zaman damgaları gibi).
AI destekli log analizi hakkındaki yazımda [buradan](https://furkanikkan.com/urun/ai-destekli-log-analizi-syslog-tan-gecici-iptables-kurali-uretme-101) da belirttiğim gibi, gözlemlenebilirlik kritik — watermark gecikmesi ve güncelleme sıklığı üzerindeki metriklerle birlikte kullanarak erken regresyonları yakala.
Kapak görseli: Unknown · PDM (Wikimedia Commons / kamu malı) · https://commons.wikimedia.org/wiki/File:Austrian_stock_market_index.jpg