Flink + Postgres: akan veriyi yere indirelim

Musluktan Kovaya: Akan Veriyi Toplama

Geçen derste musluğu açtık (Kafka olay günlüğü). Bu ders akan veriyi alıp bir yere indireceğiz: bir kelime sayacı yazacağız. Akan sipariş mesajlarından en çok geçen kelimeleri sayıp Postgres tablosuna yazacağız.

İşin büyük aracı Apache Flink. Python’da yazılmış hali PyFlink. Üç parçadan oluşuyor:

1. Source: Redpanda / Kafka’dan akan veriyi oku.
2. Transform: Kelimelere ayır, say.
3. Sink: Postgres’e yaz.

Flink Job’un Yapısı

PyFlink’te her iş bir StreamExecutionEnvironment ile başlar. Bir env (ortam) oluşturur, kaynak-bağlantı-dönüşüm-hedef zincirini kurarsın, sonra execute() ile çalıştırırsın.

from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)              # tek iş parçacığında çalışsın (öğrenci kolaylığı)

Burada parallelism=1 diyoruz: tek bir iş parçacığında çalışsın. Gerçek projede makinenin çekirdek sayısına göre 4, 8, 16 yaparsın, Flink paralel çalışır.

Source: Redpanda’dan Akan Veri

Redpanda, Kafka ile aynı protokolü konuşur (Kafka API uyumlu). Kaynak bağlantısı KafkaSource ile yapılır:

from pyflink.datastream.connectors.kafka import KafkaSource
from pyflink.common.serialization import SimpleStringSchema

source = KafkaSource.builder() \
    .set_bootstrap_servers("localhost:9092") \
    .set_topics("siparisler") \
    .set_group_id("kelime-sayaci") \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

akış = env.from_source(source, watermark_strategy=None, source_name="siparisler")

Burada set_topics("siparisler") dinlemek istediğimiz topic. set_value_only_deserializer(SimpleStringSchema()) mesajın gövdesini düz metin olarak al. Flink aynı anda key ve value ile de çalışabilir ama biz basit tutuyoruz.

Transform: Kelimelere Ayır ve Say

Gelen her mesaj bir cümle gibi (ör. “limonata çilek buz”). Onu kelimelere bölüp her kelime için (kelime, 1) üreteceğiz. Sonra Flink’in pencere fonksiyonuyla 30 saniyelik pencerelerde sayacağız.

from pyflink.datastream.functions import MapFunction, ReduceFunction
from pyflink.common.typeinfo import Types

class KelimelereAyir(MapFunction):
    def map(self, deger):
        # her kelime için ayrı kayıt üret
        kelimeler = deger.lower().split()
        return [(k, 1) for k in kelimeler]

class Topla(ReduceFunction):
    def reduce(self, a, b):
        # (kelime, adet) çiftlerini topla
        return (a[0], a[1] + b[1])

# düz metin -> (kelime, 1) çiftleri
cevirilmis = (
    akış
    .flat_map(KelimelereAyir())
    .returns(Types.TUPLE([Types.STRING(), Types.INT()]))
    .key_by(lambda x: x[0])                 # kelimeye göre grupla
    .reduce(Topla())
)

flat_map her mesajdan birden çok kayıt üretir (burada her kelime için bir tane). key_by kelimeye göre gruplar (Flink aynı anahtarlı kayıtları aynı iş parçacığına yönlendirir). reduce aynı anahtarlı kayıtları birleştirir; burada sayaç birikir.

Sink: Postgres’e Yazılsın

Hesaplanan (kelime, adet) çiftlerini bir Postgres tablosuna yazmak için JDBC sink kullanırız:

from pyflink.datastream.connectors.jdbc import JdbcSink
from pyflink.api.common.serialization import Encoder

# Postgres'e yazılacak SQL
sorgu = """
INSERT INTO kelime_sayilari (kelime, adet)
VALUES (?, ?)
ON CONFLICT (kelime) DO UPDATE SET adet = EXCLUDED.adet
"""

sink = JdbcSink.sink(
    sorgu,
    JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .with_url("jdbc:postgresql://localhost:5432/elif")
        .with_driver_name("org.postgresql.Driver")
        .with_username("elif")
        .with_password("gizli")
        .build(),
    JdbcExecutionOptions.builder()
        .with_batch_size(100)               # 100 kayıt biriktir, toplu yaz
        .with_batch_interval_ms(2000)       # veya 2 saniyede bir
        .build(),
)

cevirilmis.add_sink(sink)

ON CONFLICT (kelime) DO UPDATE kısmı upsert: aynı kelime zaten varsa sayıyı günceller, yoksa yeni satır ekler. Bu sayede tablo gittikçe büyür, “limonata” her geldiğinde sayacı artar.

JDBC jar dosyasını Flink’e göstermeyi unutma: indirilen postgresql-42.7.1.jar dosyasını env.add_jars() ile ekle veya --jar ile geç.

İşi Çalıştır

env.execute("kelime-sayaci-job")

Çıktı (terminalde):

SLF4J: ... başlatıldı
... kelime-sayaci-job iş başladı
... Subtask #1 veri akıyor

Postgres tarafında:

SELECT * FROM kelime_sayilari ORDER BY adet DESC LIMIT 5;

 kelime   | adet
----------+------
 limonata |  412
 nane     |  298
 çilek    |  187
 buzlu    |   94
 klasik   |   78

Elif’in patronu artık her gün “en çok sipariş edilen içecek” listesini anlık görebilir.

Özet

Bugün öğrendik:

• PyFlink = akan veri işleme motoru. Kafka / dosya / soket kaynak olur, JDBC / dosya / Kafka hedef olur.
• flat_map her mesajdan birden çok kayıt üretir; burada kelimelere ayırdık.
• key_by + reduce ile akan veride sürekli toplama yaparsın (kelime sayaç gibi).
• JdbcSink sonucu Postgres’e yazar; upsert ile aynı anahtar güncellenir.

Küçük Ödev

1. Yukarıdaki job’u çalıştır. Postgres’e 5 farklı kelime içeren bir mesaj gönder, tabloyu kontrol et.
2. with_batch_size(100) yerine (1) yap. Ne değişir? Bir cümle yaz.
3. Bonus: key_by(lambda x: x[0]) yerine ilk 3 harfe göre grupla (x[0][:3]). Ne gözlemledin?

Sıradaki

Modül 7 bitti. Şimdi tüm öğrendiklerini birleştirme zamanı: capstone proje geliyor →