Apache Flink (Streaming) — DE Interview Prep(Streaming) — Подготовка к DE-интервью

What a watermark actually isЧто такое watermark на самом деле

A marker flowing in the stream: W(t) = "no event with timestamp ≤ t will arrive after this." It is Flink's notion of event-time progress and is what fires event-time windows (every window whose end ≤ W).

Маркер, текущий в потоке: W(t) = «событий с меткой времени ≤ t после этого момента не будет». Это понятие прогресса event-time в Flink и то, что запускает event-time окна (каждое окно, конец которого ≤ W).

// bounded out-of-orderness: watermark = maxSeenTs - 5s
WatermarkStrategy
  .forBoundedOutOfOrderness(Duration.ofSeconds(5))
  .withTimestampAssigner((event, ts) -> event.getEventTime())
  .withIdleness(Duration.ofMinutes(1)); // don't let idle partition stall WM

• Per-partition watermarks: operator watermark = minimum across all inputs/partitions. Slowest input gates progress.
• Idle source stalls the min forever → withIdleness() marks it idle.
• The knob: bigger out-of-orderness = more correctness for late data, more latency. Smaller = faster, more drops.

• Watermarks по партициям: watermark оператора = минимум по всем входам/партициям. Самый медленный вход блокирует прогресс.
• Idle source блокирует минимум навсегда → withIdleness() помечает его idle.
• Ручка настройки: больше out-of-orderness = больше корректности для поздних данных, больше задержка. Меньше = быстрее, больше сброшенных данных.

Machinery: Assigner (which window) · Trigger (when to fire) · Evictor (optional removal) · allowed lateness.

Механизм: Assigner (какое окно) · Trigger (когда запускать) · Evictor (опциональное удаление) · allowed lateness.

stream.keyBy(e -> e.userId())
  .window(TumblingEventTimeWindows.of(Time.minutes(1)))
  .allowedLateness(Time.seconds(30))
  .sideOutputLateData(lateTag)   // don't drop late data silently
  .aggregate(new CountAgg());

Late data handlingОбработка опоздавших данных

• Past the watermark = late. Default: dropped.
• Allowed lateness: keep window state open longer; late events re-fire / update result.
• Side output (OutputTag): route too-late events to a separate stream instead of losing them.

• Позже watermark = опоздало. По умолчанию: сброшено.
• Allowed lateness: держать состояние окна открытым дольше; опоздавшие события перезапускают / обновляют результат.
• Side output (OutputTag): направить слишком поздние события в отдельный поток вместо потери.

Rescaling & TTLМасштабирование и TTL

• Key groups = atomic unit of keyed-state redistribution. Change parallelism → key groups reassigned across subtasks.
• Max parallelism fixes the number of key groups → caps how far you can rescale without a state migration. Set it deliberately at job creation.
• State TTL (StateTtlConfig): expire old keys or state grows forever → slow recovery, disk blowup. Essential at per-user, billions-of-keys scale.

• Key groups = атомарная единица перераспределения keyed-state. Изменение parallelism → key groups переназначаются между subtasks.
• Max parallelism фиксирует число key groups → ограничивает, насколько можно масштабироваться без миграции состояния. Устанавливай осознанно при создании джоба.
• State TTL (StateTtlConfig): устаревание старых ключей, иначе состояние растёт вечно → медленное восстановление, взрыв диска. Критично на масштабе миллиардов ключей по пользователям.

Workflow: stop job with savepoint → deploy new code → resume from savepoint. State survives a deploy as long as schema + UIDs are compatible.

Workflow: остановить джоб с savepoint → задеплоить новый код → продолжить из savepoint. Состояние переживает деплой, пока схема + UID совместимы.

How exactly-once worksКак работает exactly-once

• Exactly-once = state consistency, NOT "each message on the wire once." Every record affects state exactly once, even across failures.
• Mechanism: asynchronous barrier snapshotting (Chandy-Lamport variant).
• Coordinator injects checkpoint barriers into sources; barriers flow with data; on barrier, operator snapshots state.
• Barrier alignment: multi-input op waits for the barrier on ALL inputs (buffering fast ones) before snapshotting → this is what guarantees exactly-once. At-least-once skips alignment → faster but may double-count.
• Unaligned checkpoints: let barriers overtake in-flight data under backpressure (stores in-flight buffers in snapshot) → fast checkpoints when backpressured.

• Exactly-once = согласованность состояния, НЕ «каждое сообщение на проводе один раз». Каждая запись влияет на состояние ровно один раз, даже при сбоях.
• Механизм: асинхронное барьерное снапшотирование (вариант Chandy-Lamport).
• Координатор инжектит checkpoint barriers в источники; барьеры текут с данными; при барьере оператор делает снапшот состояния.
• Barrier alignment: оператор с несколькими входами ждёт барьера на ВСЕХ входах (буферизуя быстрые) перед снапшотом → это и гарантирует exactly-once. At-least-once пропускает alignment → быстрее, но может задвоить.
• Unaligned checkpoints: пропускают барьеры обгонять in-flight данные при backpressure (сохраняют in-flight буферы в снапшоте) → быстрые checkpoints при backpressure.

Guarantee ladder & costЛестница гарантий и цена

Runtime componentsRuntime-компоненты

• JobManager (master): schedules, coordinates checkpoints, handles failover. Contains Dispatcher, ResourceManager, per-job JobMaster.
• TaskManagers (workers): run subtasks in task slots (resource isolation). Slot sharing lets a pipeline slice share a slot.
• Dataflow: program → logical JobGraph → physical ExecutionGraph (parallel subtasks). Operators fuse into operator chains (one thread, no serialization → big perf win).

• JobManager (мастер): планирует, координирует checkpoints, обрабатывает failover. Содержит Dispatcher, ResourceManager, per-job JobMaster.
• TaskManagers (воркеры): выполняют subtasks в task slots (изоляция ресурсов). Slot sharing позволяет срезу пайплайна делить slot.
• Dataflow: программа → логический JobGraph → физический ExecutionGraph (параллельные subtasks). Операторы сливаются в operator chains (один поток, без сериализации → большой выигрыш в производительности).

BackpressureBackpressure

• Downstream can't keep up → it slows upstream via credit-based flow control (receiver grants buffer credits) instead of dropping/OOM. Pressure propagates back to source.
• Diagnose in Web UI: bottleneck = first operator that is busy but not backpressured.
• Causes: key skew, slow sink, GC, undersized parallelism. Heavy backpressure delays barriers → slow checkpoints (→ unaligned helps).

• Downstream не успевает → замедляет upstream через credit-based flow control (получатель выдаёт кредиты на буфер) вместо сброса/OOM. Давление распространяется к источнику.
• Диагностика в Web UI: узкое место = первый оператор, который занят, но не под backpressure.
• Причины: перекос по ключам, медленный sink, GC, недостаточный parallelism. Сильный backpressure задерживает барьеры → медленные checkpoints (→ unaligned помогает).

Partitioning & low-level APIsПартиционирование и низкоуровневые API

• keyBy hash partition (same key→same subtask) · rebalance round-robin (fix skew) · rescale local round-robin · broadcast to all (broadcast-state pattern for dynamic rules) · forward 1:1 (enables chaining).
• KeyedProcessFunction: lowest-level — events + keyed state + event/processing-time timers. For custom state machines, dedup, timeouts, joins.
• FlinkCEP: pattern detection ("A then B within 10m, not preceded by C") via NFA — fraud/abuse/alerting.
• Stream joins: interval join (bounded), CoProcessFunction (manual TTL), temporal table join (as-of dimension). Always bound join state.

• keyBy хеш-партиционирование (тот же ключ→тот же subtask) · rebalance round-robin (лечить перекос) · rescale локальный round-robin · broadcast всем (broadcast-state паттерн для динамических правил) · forward 1:1 (позволяет chaining).
• KeyedProcessFunction: самый низкоуровневый — события + keyed state + event/processing-time таймеры. Для кастомных state machines, dedup, таймаутов, join.
• FlinkCEP: обнаружение паттернов («A затем B в пределах 10m, не предшествует C») через NFA — фрод/абьюз/алертинг.
• Stream joins: interval join (ограниченный), CoProcessFunction (ручной TTL), temporal table join (as-of измерение). Всегда ограничивай join-состояние.

One-liner: Flink for true low-latency, large-stateful, event-time-correct streaming. Spark SS when seconds are fine and you want one engine for batch + stream + ML on an existing stack.

Одной строкой: Flink для настоящего низкозадержечного, large-stateful, event-time-корректного стриминга. Spark SS, когда секунды приемлемы и нужен один движок для батча + потока + ML на существующем стеке.

Iceberg + the canonical pipelineIceberg и канонический пайплайн

• Flink reads/writes Iceberg (stream + batch); streaming writes commit as Iceberg snapshots aligned with Flink checkpoints → exactly-once into a lakehouse.
• Iceberg = open table format giving ACID snapshot isolation, schema evolution, time travel over object storage. Open-source analog to Snowflake/BigQuery/Hive formats you've used.

• Flink читает/пишет Iceberg (поток + батч); потоковые записи коммитятся как Iceberg-снапшоты, выровненные с Flink checkpoints → exactly-once в lakehouse.
• Iceberg = открытый табличный формат, дающий ACID snapshot isolation, schema evolution, time travel поверх объектного хранилища. Открытый аналог форматов Snowflake/BigQuery/Hive, которые ты использовал.

Top production gotchasГлавные подводные камни в продакшене

• No .uid() → can't restore from savepoint after a code edit.
• No state TTL → state grows forever, recovery gets slower over time.
• Idle Kafka partition → min watermark stalls → windows never fire. Use withIdleness.
• Non-idempotent sink → exactly-once is a lie; you double-write on replay.
• Key skew → one subtask backpressured → checkpoints time out. Salt / two-phase aggregate.
• Dropping late data silently → use side outputs to capture it.
• Unbounded stream join → state explosion. Use interval/temporal joins or TTL.

• Нет .uid() → не восстановить из savepoint после правки кода.
• Нет state TTL → состояние растёт вечно, восстановление замедляется со временем.
• Idle партиция Kafka → min watermark застревает → окна не запускаются. Использовать withIdleness.
• Не идемпотентный sink → exactly-once — ложь; ты задваиваешь запись при повторе.
• Перекос по ключам → один subtask под backpressure → checkpoints таймаутятся. Salt / двухфазная агрегация.
• Тихий сброс опоздавших данных → использовать side outputs, чтобы захватить их.
• Неограниченный stream join → взрыв состояния. Использовать interval/temporal join или TTL.

Drill: "Checkpoints keep timing out under load."Разбор: «Checkpoints таймаутятся под нагрузкой»

1. Web UI: is it alignment time (backpressure) or snapshot time (state size)?
2. Alignment-bound → find bottleneck (busy, not backpressured); fix skew / scale / speed sink; try unaligned checkpoints.
3. State-size-bound → incremental checkpoints (RocksDB) + state TTL; relax interval/timeout.
4. Sink-bound (slow 2PC commit) → investigate external latency; idempotent sink may beat 2PC.
5. Confirm with metrics, not vibes.

1. Web UI: это alignment time (backpressure) или snapshot time (размер состояния)?
2. Ограничен alignment → найти узкое место (занят, но не под backpressure); чинить перекос / масштабировать / ускорить sink; попробовать unaligned checkpoints.
3. Ограничен размером состояния → инкрементальные checkpoints (RocksDB) + state TTL; ослабить интервал/таймаут.
4. Ограничен sink (медленный 2PC commit) → исследовать внешнюю задержку; идемпотентный sink может быть лучше 2PC.
5. Подтверждать метриками, а не ощущениями.

When NOT to use FlinkКогда НЕ использовать Flink

• Pure batch ETL where seconds-minutes latency is fine and the org is on Spark.
• Low volume where a consumer or cron suffices — Flink's ops overhead isn't justified.
• No JVM/streaming ops capacity — exactly-once + state + savepoint discipline is real cost.
• Picking boring batch when realtime isn't needed is senior judgment.

• Чистый батч ETL, где задержка в секундах-минутах приемлема, и организация на Spark.
• Малый объём, где хватает консьюмера или cron — операционные расходы Flink не оправданы.
• Нет JVM/streaming-ops компетенций — exactly-once + состояние + savepoint дисциплина — реальная цена.
• Выбирать скучный батч, когда realtime не нужен — это senior-суждение.