Spark Structured Streaming ile Akış Verisi İşleme

Structured Streaming nedir, nasıl çalışır?

Structured Streaming, Apache Spark'ın SQL motoru üzerine kurulu akış işleme modelidir. Temel fikir şudur: akan veri, sürekli yeni satırlar eklenen bir girdi tablosu olarak düşünülür ve sorgunuz bu tablo üzerinde çalışır. Spark belgelerinin anlattığı gibi, aynı DataFrame ve Dataset API'leriyle hem sınırlı (toplu) hem sınırsız (akış) veri temsil edilebilir; bu yüzden toplu işleme yazmayı bilen biri akış sorgusuna kolayca geçebilir.

Varsayılan çalışma biçimi mikro gruptur: sorgu, önceki grup işlenmeyi bitirir bitirmez yeni gelen veriden bir grup oluşturur. İsterseniz grupların sabit aralıklarla başlatılacağı bir tetikleme tanımlayabilirsiniz. Her tetiklemede sonuç tablosu güncellenir ve hedefe yazılır.

Bu modelin önemli sonucu, akış sorgusu bir kez yazıldığında sürekli çalışmasıdır. Siz tek bir sorgu tanımlarsınız; Spark yeni veri geldikçe onu artımlı olarak işler. Akış işlemenin genel kavramı için akış işleme sözlük sayfasına, Spark'ın toplu işleme tarafı için PySpark ile büyük veri işleme sayfasına bakabilirsiniz.

Kaynak ve hedef: sorguyu nasıl kurarsınız?

Bir akış sorgusu üç parçadan oluşur: okuma, dönüşüm ve yazma. Okuma readStream, yazma writeStream ile yapılır.

Belgelerde yer alan yerleşik kaynaklar şunlardır: dizine yazılan dosyaları akış olarak okuyan dosya kaynağı, Kafka kaynağı, yalnızca test için önerilen soket kaynağı ve test ile kıyaslama için hız (rate) kaynağı. Hedeflerde dosya, Kafka, konsol ve bellek hedefleri bulunur; konsol ve bellek hedefleri geliştirme ve hata ayıklama içindir.

Bir Kafka kaynağından okuyup dosyaya yazan sorgunun iskeleti şöyledir:

• olaylar = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "<sunucu:port>").option("subscribe", "<konu>").load()
• sonuc = olaylar.selectExpr("CAST(value AS STRING) AS ham")
• sorgu = sonuc.writeStream.format("parquet").option("path", "<hedef-dizin>").option("checkpointLocation", "<checkpoint-dizini>").start()

Kafka'dan gelen kayıtta değer sütunu ikili (binary) olduğundan, JSON gibi bir biçime ayrıştırmadan önce metne çevirmek gerekir; gerçek bir projede burada bir şema tanımlanır. Dikkat edilecek nokta, start() çağrılana kadar hiçbir şeyin çalışmamasıdır; start bir StreamingQuery nesnesi döndürür ve sorgu arka planda sürekli işler.

Olay zamanı ve pencereleme

Akış verisinde iki zaman vardır: olayın oluştuğu an (olay zamanı) ve sisteme ulaştığı an. Cihazdan gelen bir ölçüm ağ gecikmesiyle geç ulaşabilir. Zamana göre toplulaştırma yapacaksanız olay zamanını kullanırsınız; böylece geç gelen kayıt da doğru pencereye düşer.

Spark belgelerinde anlatılan pencere türleri dönen (tumbling), kayan (sliding) ve oturum (session) pencereleridir. Dönen pencereler sabit boyutlu ve üst üste binmeyen aralıklardır; her kayıt tek bir pencereye girer. Kayan pencerelerde pencereler üst üste binebilir, bir kayıt birden çok pencereye atanır. Oturum pencereleri ise boyutu veriye göre değişir: bir girdiyle başlar, belirlenen boşluk süresi içinde yeni girdi gelirse genişler.

Dönen ya da kayan pencere için window fonksiyonu kullanılır:

• sayim = olaylar.groupBy(window("olay_zamani", "<pencere süresi>"), "kategori").count()

Kurgusal bir örnek: bir mağaza zincirinin kasa olayları akış olarak gelir; ekip, her kategorideki satış adedini belirli aralıklı pencerelerde saymak ister. Olay zamanı kasa fişinin saatidir; ağ sorunu nedeniyle geç ulaşan fiş yine kendi saatinin penceresine eklenir.

Watermark: geç veriyi ne kadar beklersiniz?

Geç gelen veriyi sonsuza dek beklemek, motorun her pencerenin durumunu sonsuza dek bellekte tutması demektir. Watermark bunu sınırlar: motor, görülen en büyük olay zamanından belirlediğiniz gecikme süresi kadar geriyi eşik olarak alır. Eşiğin gerisinde kalan pencerelerin durumu atılır; eşikten çok geride kalan kayıtların işlenmesi garanti edilmez.

Söz dizimi şöyledir ve toplulaştırmadan önce çağrılması gerekir:

• sayim = olaylar.withWatermark("olay_zamani", "<gecikme süresi>").groupBy(window("olay_zamani", "<pencere süresi>"), "kategori").count()

Spark belgeleri iki kurala dikkat çeker. Watermark, toplulaştırmada kullanılan zaman sütunuyla aynı sütun üzerinde tanımlanmalıdır. İkincisi, withWatermark toplulaştırmadan önce çağrılmalıdır; sonradan çağrılırsa append kipinde geçersiz olur.

Gecikme süresi bir ödünleşimdir. Uzun tolerans daha çok geç kaydı yakalar ama durum bilgisi büyür ve append kipinde sonuçlar geç yazılır; kısa tolerans düşük gecikme sağlar ama geç gelen kayıtları kaybedebilir. Süreyi, verinizin gerçek gecikme dağılımına bakarak seçin; tahminle koymayın.

Çıktı kipleri: append, update ve complete

Çıktı kipi, sonuç tablosunun hedefe nasıl yazılacağını belirler. Seçim hem sorgu türüne hem hedefe bağlıdır.

Append kipinde hedefe yalnızca son hali belli olan satırlar eklenir. Watermark'lı bir pencere toplulaştırmasında bir pencere, watermark o pencerenin ötesine geçtikten sonra yazılır; yani sonuçlar bir gecikmeyle gelir. Update kipinde son tetiklemeden bu yana değişen satırlar hedefe yazılır; dosya gibi ince taneli güncellemeyi desteklemeyen hedefler için uygun değildir. Complete kipinde ise sonucun tamamı her tetiklemede yazılır; bu kip tüm toplulaştırma verisinin tutulmasını gerektirir ve bu yüzden watermark ile eski durumu atmaya izin vermez.

Pratik bir kural: dosya hedefi ve watermark'lı pencere sayımı için append, küçük bir toplulaştırma tablosunu sürekli güncellemek için update ya da complete uygundur. Geliştirme sırasında konsol hedefiyle farklı kipleri deneyerek davranışı gözlemek öğreticidir.

Checkpoint ve hatadan devam

Akış sorgusu uzun süre çalıştığından hata ya da yeniden başlatma kaçınılmazdır. Checkpoint, sorgunun kaldığı yerden devam etmesini sağlar. Spark belgelerine göre sorgu, her tetiklemede işlenen ofset aralığı gibi ilerleme bilgisini ve çalışan toplulaştırma durumunu checkpoint konumuna kaydeder. Bu konum HDFS uyumlu bir dosya sisteminde bir yol olmalıdır ve writeStream üzerinde checkpointLocation seçeneğiyle verilir.

Birkaç pratik not: her sorgunun kendi checkpoint konumu olmalıdır, iki sorgu aynı konumu paylaşmamalıdır. Checkpoint konumunu silmek, sorguyu baştan başlatmak anlamına gelir. Sorgunun kodu değiştirildiğinde, değişikliğin checkpoint ile uyumlu olup olmadığı belgelerdeki 'Recovery Semantics after Changes in a Streaming Query' bölümünde anlatılır; durumu etkileyen değişikliklerde yeni bir checkpoint konumu gerekebilir.

Üretime alırken sorgunun sağlığını izleyin: işlenen satır hızı, tetikleme süresi ve girdi hızı arasındaki fark, sorgunun yetişip yetişmediğini gösterir. Girdi hızı sürekli işlem hızını aşıyorsa gecikme birikir.

Sık hatalar

En sık hata, olay zamanı yerine işleme zamanını kullanmaktır; geç gelen kayıtlar yanlış pencereye girer. Doğrusu, kaynak veride olay zamanı sütununu belirlemek ve pencereyi onun üzerinden kurmaktır.

İkinci hata, watermark'ı toplulaştırmadan sonra ya da farklı bir sütunda tanımlamaktır. Üçüncü hata, watermark'sız toplulaştırmada durumun sürekli büyüyeceğini unutmaktır; pencere sayımlarında durum yönetimi watermark ile sınırlanmalıdır. Dördüncü hata, soket ve hız gibi test kaynaklarını üretime taşımaktır; belgeler bunların uçtan uca hata toleransı sağlamadığını belirtir. Beşinci hata, checkpoint konumunu geçici bir diskte tutmaktır; konum kalıcı bir dosya sisteminde olmalıdır.

Bu beceriyi Mindset programlarında nerede çalışırsınız?

Akış verisi işleme, veri mühendisleri ve gerçek zamanlı analitik ekipleri için, sipariş akışı, sensör verisi ya da kullanıcı olayı gibi sürekli gelen veriyi anlamlı sonuca çevirmenin yoludur.

Mindset'in Spark Structured Streaming ile Akış İşleme programı Structured Streaming modelini, kaynak ve hedefleri ve şema yönetimini, zaman, pencereleme ve geç veriyi, durum bilgili işlemeyi, hata toleransı ve üretime almayı ele alır; performans ve izleme ile bir bitirme projesi de kapsamındadır.