It is 2:14 AM at a company like PhonePe. A UPI payment lands. The payer's balance has to update, the payee has to get notified, and a fraud check has to run - all inside 500 milliseconds. Your nightly Airflow DAG that processes yesterday's transactions in a neat batch job cannot help here. The payment has already been approved or declined by the time that DAG even wakes up. This is the gap Flink fills. It does not replace your batch pipelines - it handles the class of problems where "wait until tonight" is not an acceptable answer. **Apache Flink** is a distributed engine built to process data as an endless stream of events, computing results continuously instead of waiting for a batch window to close. Think of a batch job like a photograph - it captures one moment and processes everything up to that point. Flink is closer to a live video feed - it keeps producing results as new events arrive, forever, without a defined end. > 📌 **Remember:** Streaming is not "batch, but faster." It is a different > processing model - results appear continuously as data flows in, not as one > big compute pass over a fixed dataset. As covered in the Kafka module, Kafka's job is durable, ordered delivery of events. Flink's job is different - it reads those events and actually computes something over them: running totals, fraud scores, alerts, aggregated windows. Kafka moves the data. Flink does the thinking. ### What this module covers, and what it deliberately does not This module gives you a **working understanding** of stream processing through Flink SQL - enough to read a Flink job, reason about what it is doing, and write a simple windowed aggregation yourself. It is not Flink operational mastery. Out of scope, named on purpose so you know they exist as a later step: the low-level DataStream API, custom operators, deep state backend tuning, advanced checkpoint configuration, production cluster deployment, and the full connector catalog. If you end up owning a Flink cluster in production, you will learn those on the job, or in a dedicated deep-dive later in your career. ### A quick self-check before moving on * In one sentence, why can't a nightly batch job power real-time fraud detection? * What is the core difference between what Kafka does and what Flink does? ---
Flink actually gives you three ways to write jobs: the low-level **DataStream API** (Java/Scala/Python, full control over every operator), the **Table API** (a DataFrame-style API), and **Flink SQL** (plain SQL over streaming tables). > 📌 **Engineering Decision:** For most data engineering ETL-style streaming > jobs - windowed aggregations, filtering, simple joins, writing results to a > sink - Flink SQL is the right tool, not the DataStream API. It is far easier > to read, easier for a teammate to maintain six months later, and it is what > this module deliberately focuses on. Reach for the DataStream API only when > you need custom logic that genuinely cannot be expressed as SQL - complex > per-event state machines, custom timers, or non-relational processing. Most > data engineers will never need to open that door. This mirrors a pattern you already know: just like you would rather write a dbt model in SQL than a hand-rolled Python transformation when SQL can express the same logic clearly, Flink SQL should be your default, with the DataStream API as the escape hatch. ```sql -- Flink SQL: read from a Kafka topic and count events per merchant every 5 minutes SELECT window_start, window_end, merchant_id, COUNT(*) AS transaction_count FROM TABLE( TUMBLE(TABLE upi_transactions, DESCRIPTOR(event_time), INTERVAL '5' MINUTES) ) GROUP BY window_start, window_end, merchant_id; ``` > **Note:** `TABLE(TUMBLE(...))` is a table-valued function - it takes the > stream and a window definition and returns a table you can `GROUP BY` just > like any other SQL table. You do not need to memorize the syntax right now, > just recognize the shape: it looks like ordinary SQL with one extra function. ---
A Flink SQL job that reads from Kafka and writes results somewhere follows the same three-step shape every time: declare a source table, declare a sink table, then run a query that reads from one and writes to the other. +-------------+ +--------------+ +-------------+ | Kafka | ---> | Flink SQL | ---> | Sink table | | topic | | windowed | | (S3/Delta) | | (source) | | aggregation | | | +-------------+ +--------------+ +-------------+ This is the same shape you saw in the Kafka module's Kafka Connect section - data moves from one system to another - except here, real computation happens in the middle instead of a pass-through copy. ```sql -- Step 1: declare the Kafka topic as a source table CREATE TABLE upi_transactions ( transaction_id STRING, merchant_id STRING, amount DECIMAL(10, 2), event_time TIMESTAMP(3), -- tells Flink which column is the event's real-world timestamp WATERMARK FOR event_time AS event_time - INTERVAL '10' SECONDS ) WITH ( 'connector' = 'kafka', 'topic' = 'upi-transactions', 'properties.bootstrap.servers' = 'kafka-broker-mumbai:9092', 'format' = 'avro-confluent' ); -- Step 2: declare where results should land CREATE TABLE merchant_volume_5min ( window_start TIMESTAMP(3), window_end TIMESTAMP(3), merchant_id STRING, total_amount DECIMAL(12, 2) ) WITH ( 'connector' = 'filesystem', 'path' = 's3://prod-mumbai-lake/gold/merchant_volume_5min', 'format' = 'parquet' ); -- Step 3: run the windowed aggregation INSERT INTO merchant_volume_5min SELECT window_start, window_end, merchant_id, SUM(amount) AS total_amount FROM TABLE( TUMBLE(TABLE upi_transactions, DESCRIPTOR(event_time), INTERVAL '5' MINUTES) ) GROUP BY window_start, window_end, merchant_id; ``` > **Note:** The `avro-confluent` format connects this job to the Schema Registry > from the Kafka module - Flink reads the same Avro schema your producers > registered, so a field rename upstream breaks this job the same way it would > break any other consumer. > 💡 **Practice:** Using the query above as a template, change the window from > 5 minutes to 1 hour and rename the output table to `merchant_volume_hourly`. > What do you expect to happen to the row count in the output compared to the > 5-minute version, and why? ---
A stream never ends, so "give me the total" is a meaningless question on its own - total of what, up to when? **Windows** are how you carve an infinite stream into finite, computable chunks. Flink supports several window types; this module covers the three you will actually reach for. **Tumbling windows** are fixed-size, back-to-back, non-overlapping slices of time. Every event belongs to exactly one window. This is what the query above uses - a clean 5-minute bucket, then the next 5-minute bucket, with no overlap and no gaps. Use tumbling windows for periodic reports: hourly transaction counts, daily active users, five-minute merchant volume. Tumbling windows (5 min each, no overlap) |--- 00:00-00:05 ---|--- 00:05-00:10 ---|--- 00:10-00:15 ---| window 1 window 2 window 3 **Sliding windows** overlap on purpose. A 10-minute sliding window that advances every 1 minute means every event can appear in up to 10 different windows. Use sliding windows when you want a smoothed trend line instead of discrete buckets - a rolling 10-minute average that updates every minute, the way a dashboard "last 15 minutes" widget usually works. **Session windows** group events by activity, not by a fixed clock. A session closes after a gap of inactivity - say, 30 minutes with no events for that key - then a new session starts on the next event. Use session windows for user behavior analysis: how long did a customer's browsing session last on the Swiggy app before they either ordered or gave up. | Window type | Boundary rule | Good for | |:---|:---|:---| | Tumbling | Fixed size, no overlap | Periodic reports (hourly, daily) | | Sliding | Fixed size, overlapping | Rolling/trend metrics | | Session | Gap of inactivity | User activity/behavior analysis | > 🔴 **Common Mistake:** Reaching for a sliding window when a tumbling window > would do. Sliding windows are more expensive - Flink keeps more overlapping > state alive per key - because every event lives in multiple windows at once. > If you just need "totals per hour," a tumbling window is simpler, cheaper, > and easier to reason about. Save sliding windows for when you genuinely need > a rolling trend. ### Concept check * You need "orders per 10-minute block, no overlap, for a daily report." Which window type fits? * You need "how long did each user stay active in the app before going idle for 20 minutes." Which window type fits? ---
Every event has two possible timestamps: **event time**, when it actually happened in the real world, and **processing time**, when Flink happens to see it. These are rarely the same moment. A mobile payment made on a Bengaluru metro with patchy connectivity might reach your Kafka topic three minutes after it actually occurred. > 📌 **Remember:** Always use event time for business-meaningful windows like > "revenue per hour." If you use processing time instead, a spike in network > latency silently shifts transactions into the wrong hour's total, and nobody > notices until a monthly reconciliation report doesn't add up. So how does Flink know when a 5-minute window is "done," if events for that window can still be arriving late? This is exactly what a **watermark** answers. A watermark is Flink's running estimate of how far event time has progressed - not a hard promise about lateness, but a signal that says "events older than this point are increasingly unlikely to still show up." In the source table above, this line does the work: ```sql WATERMARK FOR event_time AS event_time - INTERVAL '10' SECONDS ``` This tells Flink to generate its watermark 10 seconds behind the most advanced event time it has observed so far. Once the watermark passes a window's end time, Flink treats that window as complete enough to close and emit - anything that arrives after the watermark has already passed is considered late. > **Note:** Resist the mental model of "10 seconds = the maximum lateness > Flink allows, guaranteed." The watermark tracks event-time progress, and > that progress depends on what events Flink has actually seen. If a Kafka > partition goes quiet, the watermark on that partition stops advancing too - > see the troubleshooting scenario below for exactly this case. Watermark example - 5 min tumbling window, 10s allowed lateness Window [00:00 - 00:05) closes when the watermark passes 00:05:00 An event timestamped 00:04:58 arriving at processing time 00:05:03 still lands inside the window correctly, because the watermark has only reached 00:04:53 at that point (10s behind real time) An event timestamped 00:04:58 arriving at processing time 00:05:12 arrives AFTER the watermark has passed 00:05:00 - it is late, and by default it is dropped from that window's result > 🔴 **Common Mistake:** Assuming events always arrive in order and skipping > the watermark configuration entirely. Real networks, retries, and mobile > clients guarantee some events arrive late. Without a watermark, Flink has no > way to know a window is "done" - either it waits forever and never emits a > result, or (with a default strategy) it uses only processing time and quietly > miscounts events that arrived out of order. ### Worked example: a late event Say your window is `[00:00, 00:05)` with a 10-second watermark delay. A transaction actually happened at `00:04:58` (event time) but reached Kafka late, arriving at `00:05:15` (processing time) because of a mobile network retry. With the watermark configured, Flink is still willing to accept events up to 10 seconds behind its watermark - so as long as the watermark itself hasn't yet advanced past `00:05:08`, this event still lands correctly in the `[00:00, 00:05)` window's total. Without any watermark strategy, Flink has no rule to apply lateness tolerance at all, and correctness depends entirely on ordering you cannot guarantee. > 💡 **Practice:** Using the source table definition above, change the > watermark delay from 10 seconds to 2 minutes. What trade-off did you just > make? (Hint: think about how long Flink now waits before it is willing to > close and emit a window.) ---
The job below was working fine for weeks. It has a properly configured watermark. But this morning, one merchant's windows have simply stopped appearing in the output, while every other merchant's windows still emit normally. Find the problem. ```sql CREATE TABLE upi_transactions ( transaction_id STRING, merchant_id STRING, amount DECIMAL(10, 2), event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '10' SECONDS ) WITH ( 'connector' = 'kafka', 'topic' = 'upi-transactions', 'properties.bootstrap.servers' = 'kafka-broker-mumbai:9092', 'format' = 'avro-confluent' ); INSERT INTO merchant_volume_5min SELECT window_start, window_end, merchant_id, SUM(amount) FROM TABLE( TUMBLE(TABLE upi_transactions, DESCRIPTOR(event_time), INTERVAL '5' MINUTES) ) GROUP BY window_start, window_end, merchant_id; ``` **What's wrong:** The Kafka topic has multiple partitions, and Flink computes a single overall watermark by taking the *minimum* watermark across every partition it reads from. One partition - say the one this quiet merchant's events happen to land on - has gone idle. No new events means no new timestamps means that partition's watermark stops advancing entirely. Because the overall watermark can only be as fresh as its slowest partition, an idle partition silently holds back window results for every key, not just the one on that partition, once it stalls for long enough. Partition 0: events keep flowing -> watermark keeps advancing Partition 1: goes idle -> watermark frozen at last seen event Overall watermark = MIN(partition 0 watermark, partition 1 watermark) Frozen partition 1 holds the whole job's watermark back This is a real, well-known Flink behavior called **watermark idleness**, and it is a far more common production issue than a missing watermark clause - a missing watermark would typically surface as a planner error at deployment time, not a silent runtime stall. **The fix:** configure an idle-source timeout so Flink stops waiting on partitions that have gone quiet, and lets the watermark advance based on the partitions that are still active: ```sql CREATE TABLE upi_transactions ( transaction_id STRING, merchant_id STRING, amount DECIMAL(10, 2), event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '10' SECONDS ) WITH ( 'connector' = 'kafka', 'topic' = 'upi-transactions', 'properties.bootstrap.servers' = 'kafka-broker-mumbai:9092', 'format' = 'avro-confluent', -- treat a partition with no new events for 60s as idle, -- so it stops holding back the overall watermark 'scan.watermark.idle-timeout' = '60s' ); ``` > 🔴 **Common Mistake:** Assuming every stalled window is caused by a missing > or misconfigured watermark clause. Just as often, the watermark is > configured correctly but a source partition has gone idle. Check consumer > lag and per-partition activity before assuming the SQL itself is wrong. ---
It is 2:14 AM at a company like PhonePe. A UPI payment lands. The payer's balance has to update, the payee has to get no...
Flink actually gives you three ways to write jobs: the low-level DataStream API (Java/Scala/Python, full control over ev...
A Flink SQL job that reads from Kafka and writes results somewhere follows the same three-step shape every time: declare...
A stream never ends, so "give me the total" is a meaningless question on its own - total of what, up to when? Windows ar...
Every event has two possible timestamps: event time, when it actually happened in the real world, and processing time, w...
The job below was working fine for weeks. It has a properly configured watermark. But this morning, one merchant's windo...
Every windowed aggregation you've written so far implies Flink is remembering something between events. When a new trans...
Your Flink job has been running for six hours, holding partial window totals in memory. Then the machine it's running on...
As covered in the Kafka module, a topic's partitions are what let multiple consumers read it in parallel. Flink is one o...
In the Data Modeling module, a batch JOIN is straightforward - both tables already fully exist, so the database can just...
The examples in this module write results to an S3/filesystem sink, which is genuinely useful for learning and for some ...
You now have three tools that look similar on the surface but solve different problems: Airflow orchestrates scheduled b...
Before diving into setup commands, here is the full shape of what you're about to build: Kafka topic (upi-transactions) ...
Prerequisites: Docker and Docker Compose installed. Completion of the Kafka module's local Kafka cluster lab. This lab d...
Concept What it means Where you'll use it Flink SQL SQL over streaming tables Default choice for ETL-style streaming job...
Reaching for the DataStream API before trying Flink SQL is the most common over-engineering mistake beginners make with ...
Aligns directly with DevOps, Site Reliability (SRE), and Platform Engineering job descriptions.