Flink + Postgres: akan veriyi yere indirelim

PyFlink: Python’da Stream Motoru

Apache Flink JVM tabanlı, dağıtılmış bir stream işleme motorudur. PyFlink, Python API’sidir. Üç temel soyutlama:

1. DataStream API: Java/Scala ile tam güç, Python ile kısıtlı.
2. Table API: SQL benzeri, type-safe.
3. SQL Client: doğrudan SQL ile job yazma.

Biz DataStream API + JDBC sink kullanacağız. Pipeline: Redpanda → flat_map → key_by → reduce → Postgres.

Ortamı Kur

PyFlink Python 3.10+ ile çalışır. Pip ile kurulum:

pip install apache-flink

JDBC sink için Postgres sürücüsü gerekir:

wget https://repo1.maven.org/maven2/org/postgresql/postgresql/42.7.1/postgresql-42.7.1.jar

Bu JAR’ı Flink’e geçmek için --jar argümanı veya env.add_jars() API’si kullanılır.

StreamExecutionEnvironment

Her PyFlink job’unda bir ortam (environment) oluşturulur. Burada paralellik, checkpointleme, state backend ayarlanır:

from pyflink.datastream import StreamExecutionEnvironment, CheckpointingMode

env = StreamExecutionEnvironment.get_execution_environment()

# Checkpoint ayarla (her 10 saniyede bir state'i kaydet)
env.enable_checkpointing(10_000)
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)

# Paralellik: 4 partition'lı bir topic için 4 idealdir
env.set_parallelism(4)

# State backend: RocksDB (büyük state için)
env.set_state_backend("rocksdb")

EXACTLY_ONCE modu, checkpoint + transactional sink birleşimi ile garantilenir. Çoğu üretim Flink job’u bu modda çalışır.

Source: Redpanda / Kafka

PyFlink’in Kafka connector’ı pyflink.datastream.connectors.kafka altında:

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

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

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

KafkaOffsetsInitializer.earliest() en eski mesajdan başla demek. Üretimde committed_offsets() kullanılır (kaldığı yerden devam).

Watermark: Geç Gelen Mesajlar

Akan veride geç gelen mesajlar (late events) sorun olur. Mesaj üretildiği saat 14:00, sisteme 14:05’te geldi. Pencere hesabı 14:00-14:05’i çoktan kapattı.

Flink’in çözümü: watermark. Bir olayın event time’ı + belirli bir gecikme toleransı = watermark. Mesajlar watermark’ın gerisindeyse pencere tekrar açılır.

from pyflink.common import WatermarkStrategy, Duration

ws = (
    WatermarkStrategy
    .for_bounded_out_of_orderness(Duration.of_seconds(10))
    .with_timestamp_assigner(lambda event, ts: extract_ts(event))
)

akış = env.from_source(source, watermark_strategy=ws, source_name="kafka-source")

Bu örnekte 10 saniyelik tolerans: 14:00’deki mesaj 14:00:10’a kadar hâlâ kabul edilir.

Transform: Kelimelere Ayır, Pencerede Say

Önce her mesajı kelimelere ayırırız:

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

class KelimelereAyir(MapFunction):
    def map(self, deger):
        # JSON string -> kelime listesi
        import json
        olay = json.loads(deger)
        cumle = (olay.get("aciklama") or "").lower()
        return [(k, 1) for k in cumle.split() if k]

cevirilmis = (
    akış
    .flat_map(KelimelereAyir())
    .returns(Types.TUPLE([Types.STRING(), Types.INT()]))
)

Şimdi 5 saniyelik tumbling pencere ile kelime başına say:

cevirilmis = (
    cevirilmis
    .key_by(lambda x: x[0])                                    # kelimeye göre grupla
    .window(TumblingEventTimeWindows.of(Duration.of_seconds(5)))
    .reduce(lambda a, b: (a[0], a[1] + b[1]))                  # pencere içinde topla
)

TumblingEventTimeWindows sabit, bitişik, örtüşmeyen pencereler: [0-5s], [5-10s], [10-15s]. Event time’a göre. Alternatifler:

• Sliding: örtüşen pencereler (5 saniyelik pencere, her 1 saniyede yeni pencere).
• Session: aktivite arasındaki boşluğa göre değişken pencere.

Sink: Postgres’e Exactly-Once

Postgres’e yazmak için JDBC sink + upsert pattern:

from pyflink.datastream.connectors.jdbc import JdbcSink, JdbcConnectionOptions, JdbcExecutionOptions

UPSERT_SQL = """
INSERT INTO kelime_sayilari (pencere_baslangici, kelime, adet)
VALUES (?, ?, ?)
ON CONFLICT (pencere_baslangici, kelime) DO UPDATE SET adet = EXCLUDED.adet
"""

# Pencere başlangıcını ekle (window olayı pencere bitiminde gelir)
class PencereBaslangiciEkle(MapFunction):
    def map(self, value):
        kelime, adet = value
        return (None, kelime, adet)            # pencere_baslangici = None (basit tutuyoruz)

sink = JdbcSink.sink(
    UPSERT_SQL,
    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(500)
        .with_batch_interval_ms(2000)
        .with_max_retries(3)
        .build(),
)

cevirilmis.map(PencereBaslangiciEkle()).returns(
    Types.TUPLE([Types.SQL_TIMESTAMP(), Types.STRING(), Types.INT()])
).add_sink(sink)

Kritik ayarlar:

• with_batch_size(500): 500 kayıt biriktir, toplu yaz. Throughput artar.
• with_batch_interval_ms(2000): 2 saniyede bir zorla yaz. Latency artar ama veri beklemez.
• ON CONFLICT … DO UPDATE: aynı pencere + kelime gelirse güncelle. Pencere olayı birden çok gelirse idempotent.

EXACTLY_ONCE ile JDBC

JDBC sink EXACTLY_ONCE’yi iki aşamalı commit (2PC) ile sağlar:

1. Flink, veritabanına “bu kayıtları yaz” der ama henüz commit etmez.
2. Checkpoint başarılı olduğunda Flink “commit” der.
3. Crash olursa 2PC kuyruğundaki işlem iptal edilir.

Postgres 2PC’yi destekler, ancak etkinleştirmek için max_prepared_transactions > 0 ayarı gerekir.

İşi Çalıştır

if __name__ == "__main__":
    env.execute("kelime-sayaci-job")

Terminal:

$ python wordcount_job.py
SLF4J: Class path contains multiple SLF4J bindings.
... kelime-sayaci-job kelime-sayaci-job iş başladı
... 4 task manager başlatıldı
... Subtask #1 veri akıyor (partition 0)
... Subtask #2 veri akıyor (partition 1)
... Subtask #3 veri akıyor (partition 2)
... Subtask #4 veri akıyor (partition 3)

Postgres tarafında her 5 saniyede yeni pencereler görünür:

SELECT pencere_baslangici, kelime, adet
FROM kelime_sayilari
WHERE pencere_baslangici > NOW() - INTERVAL '5 minutes'
ORDER BY pencere_baslangici DESC, adet DESC
LIMIT 10;

 pencere_baslangici    | kelime   | adet
-----------------------+----------+------
 2026-06-19 14:30:05   | limonata |  124
 2026-06-19 14:30:05   | nane     |   98
 2026-06-19 14:30:05   | çilek    |   67
 2026-06-19 14:30:00   | limonata |  112
 ...

Sorun Giderme

Sık karşılaşılan sorunlar:

Belirti	Olası sebep	Çözüm
ClassNotFoundException: org.postgresql.Driver	JDBC JAR yüklenmedi	--jar postgresql-42.7.1.jar ile geç
Mesajlar gelmiyor	auto.offset.reset=latest ve grup offset’i yok	earliest() veya producer’ı önce çalıştır
Aynı kayıt birden çok yazıldı	EXACTLY_ONCE aktif değil	enable_checkpointing + checkpoint mode kontrol
Checkpoint başarısız	State backend taşmış	RocksDB state backend + artımlı checkpoint
Pencere sonucu gelmiyor	Watermark ilerlemiyor	Tüm partition’larda mesaj olmalı (boş partition watermark’ı durdurur)

Özet

Bugün öğrendik:

• PyFlink = dağıtılmış stream motoru, Python API’si. EXACTLY_ONCE checkpoint ile sağlanır.
• Watermark: geç gelen mesajları belirli bir toleransla kabul etme mekanizması.
• Event-time pencere: TumblingEventTimeWindows ile sabit, örtüşmeyen pencereler.
• JdbcSink + 2PC: Postgres’e exactly-once yazma. Batch size + interval ayarı önemli.
• Operasyon: state backend RocksDB, paralellik partition sayısına eşit, JDBC JAR classpath’te.

Alıştırma

1. Yukarıdaki job’u çalıştır, Postgres tablosunu kontrol et.
2. TumblingEventTimeWindows.of(Duration.of_seconds(5)) yerine of_seconds(30) dene. Ne değişiyor?
3. Watermark’ı Duration.of_seconds(30) yap. Geç gelen bir mesajın işlendiğini gözlemle.
4. EXACTLY_ONCE’yi kapat (mode = AT_LEAST_ONCE). Postgres tablosunda tekrar sayımı gözlemle.
5. Bir cümle yaz: “Neden paralellik = topic partition sayısı?”

Sıradaki

Modül 7 bitti. Tüm öğrendiklerini bir capstone projede birleştirme zamanı →