---

name: flink description: >- Process real-time data streams with Apache Flink. Use when a user asks to build real-time analytics, process event streams with low latency, implement complex event processing, or run stateful stream processing at scale. license: Apache-2.0 compatibility: 'Java 11 or 17 runtime; Python 3.9–3.12 for PyFlink' metadata: author: terminal-skills version: 1.1.0 category: data-ai repository: https://github.com/apache/flink tags: - flink - streaming - real-time - analytics - events

Apache Flink

Overview

Flink is a distributed stream processing engine for real-time analytics. Unlike batch-first systems (Spark), Flink is stream-first — it processes events as they arrive with millisecond latency. Supports exactly-once semantics, stateful processing, and event time windowing.

This skill targets Flink 2.x. 2.3.0 is the current release, but the Kafka connector is published only for Flink 2.1 and 2.2, so the steps pin 2.2.1. Flink 2.0 removed the DataSet API, the Scala DataStream API, SourceFunction/SinkFunction and Java 8 support, so most pre-2.0 tutorials — including anything built on FlinkKafkaConsumer — no longer run.

Instructions

Step 1: PyFlink Setup

java -version                      # Java 11 or newer; 17 is the default for Flink 2.x
python3 -m venv .venv && source .venv/bin/activate  # Python 3.9–3.12
python -m pip install apache-flink==2.2.1   # newest Flink the Kafka connector supports

The pip package bundles a complete Flink distribution (CLI, cluster scripts, conf/config.yaml) but no connectors. Connectors are released separately; fetch the Kafka one from Maven Central and check it against a pinned SHA-256 (a different connector version needs its own value):

KAFKA_JAR=flink-sql-connector-kafka-5.0.0-2.2.jar
MAVEN=https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/5.0.0-2.2
mkdir -p lib
curl -fsSL -o "lib/$KAFKA_JAR" "$MAVEN/$KAFKA_JAR"
# SHA-256 of the 5.0.0-2.2 release; the command stops here if the file differs
echo "5605c691d11a501382c383fecba37a7a552467da5ab7ba904ef5d6f3d62c5616  lib/$KAFKA_JAR" | sha256sum -c -

The version is the connector release followed by the Flink minor it was built for: 5.0.0-2.2 is connector 5.0.0 for Flink 2.2.

Step 2: Stream Processing

# stream_job.py — page views per URL in 5-minute event-time windows, Kafka to Kafka
import json
from pathlib import Path

from pyflink.common import Duration, Types, WatermarkStrategy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.time import Time
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.base import DeliveryGuarantee
from pyflink.datastream.connectors.kafka import (
    KafkaOffsetsInitializer, KafkaRecordSerializationSchema, KafkaSink, KafkaSource,
)
from pyflink.datastream.window import TumblingEventTimeWindows

env = StreamExecutionEnvironment.get_execution_environment()
env.add_jars(Path("lib/flink-sql-connector-kafka-5.0.0-2.2.jar").resolve().as_uri())
env.set_parallelism(4)
env.enable_checkpointing(60_000)  # ms; state snapshots for recovery

source = (
    KafkaSource.builder()
    .set_bootstrap_servers("localhost:9092")
    .set_topics("clickstream")
    .set_group_id("flink-analytics")
    .set_starting_offsets(KafkaOffsetsInitializer.earliest())
    .set_value_only_deserializer(SimpleStringSchema())
    .build()
)
sink = (
    KafkaSink.builder()
    .set_bootstrap_servers("localhost:9092")
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
        .set_topic("page-analytics")
        .set_value_serialization_schema(SimpleStringSchema())
        .build()
    )
    .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE)
    .build()
)

# The Kafka record timestamp is the event time; tolerate 5 s of out-of-order events
watermarks = (
    WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5))
    .with_idleness(Duration.of_seconds(30))  # an empty partition must not hold windows open
)

(
    env.from_source(source, watermarks, "clickstream")
    .map(json.loads)
    .filter(lambda e: e["event_type"] == "page_view")
    .map(lambda e: (e["page_url"], 1, [e["user_id"]]))
    .key_by(lambda t: t[0])
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .reduce(lambda a, b: (a[0], a[1] + b[1], list(set(a[2] + b[2]))))
    .map(
        lambda t: json.dumps({"page_url": t[0], "view_count": t[1], "unique_users": len(t[2])}),
        output_type=Types.STRING(),
    )
    .sink_to(sink)
)

env.execute("Clickstream Analytics")

Step 3: Flink SQL

# sql_job.py — per-minute order revenue with Flink SQL
from pathlib import Path

from pyflink.table import EnvironmentSettings, TableEnvironment

t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
t_env.get_config().set(
    "pipeline.jars", Path("lib/flink-sql-connector-kafka-5.0.0-2.2.jar").resolve().as_uri()
)
# without this, source subtasks that own no Kafka partition hold every window open
t_env.get_config().set("table.exec.source.idle-timeout", "30 s")

# Kafka source table; event_time drives the watermark
t_env.execute_sql("""
    CREATE TABLE orders (
        order_id STRING,
        user_id STRING,
        amount DECIMAL(10, 2),
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'orders',
        'properties.bootstrap.servers' = 'localhost:9092',
        'properties.group.id' = 'flink-orders',
        'scan.startup.mode' = 'earliest-offset',
        'format' = 'json'
    )
""")

# Tumbling windows through the TUMBLE table-valued function
t_env.execute_sql("""
    SELECT
        window_start,
        COUNT(*) AS order_count,
        SUM(amount) AS total_revenue,
        COUNT(DISTINCT user_id) AS unique_buyers
    FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(event_time), INTERVAL '1' MINUTE))
    GROUP BY window_start, window_end
""").print()

Step 4: Submit to a Cluster

python stream_job.py runs the job in an embedded mini-cluster. To run it on a real cluster, use the scripts inside the pip package. First edit conf/config.yaml in the directory that find_flink_home.py prints ($FLINK_HOME below): under rest: uncomment bind-address: localhost, and under taskmanager: set numberOfTaskSlots: 4 (the job above needs 4 slots; the default is 1).

export FLINK_HOME="$(find_flink_home.py)"        # installed with apache-flink
"$FLINK_HOME/bin/start-cluster.sh"               # JobManager + TaskManager, web UI on http://localhost:8081
"$FLINK_HOME/bin/flink" run --detached --python stream_job.py
"$FLINK_HOME/bin/flink" list                     # running jobs with their JobIDs
"$FLINK_HOME/bin/flink" cancel 3bf2577fb9074e3a83e4fa147adfaa51
"$FLINK_HOME/bin/stop-cluster.sh"

Examples

Example 1: Page views per URL every five minutes

User: "Our site sends click events to the Kafka topic clickstream. Count page views and unique visitors per URL every five minutes and write the result to page-analytics."

Save the Step 2 code as stream_job.py and run python stream_job.py. Input messages look like {"event_type": "page_view", "page_url": "/pricing", "user_id": "u-1001"}. One message per URL and window arrives on page-analytics once the watermark passes the window end:

{"page_url": "/pricing", "view_count": 5, "unique_users": 2}
{"page_url": "/docs/quickstart", "view_count": 2, "unique_users": 1}
{"page_url": "/blog/flink-2-3", "view_count": 2, "unique_users": 1}

The newest window stays open until an event later than its end (plus the 5 s tolerance) arrives.

Example 2: Revenue per minute in SQL

User: "Show order count, revenue and distinct buyers per minute from the orders topic."

Save the Step 3 code as sql_job.py and run python sql_job.py. Each order is JSON such as {"order_id": "ord-5021", "user_id": "u-1003", "amount": 72.49, "event_time": "2026-10-01 10:03:30.000"}. The job prints a row as each window closes:

+----+-------------------------+----------------------+------------------------------------------+----------------------+
| op |            window_start |          order_count |                            total_revenue |        unique_buyers |
+----+-------------------------+----------------------+------------------------------------------+----------------------+
| +I | 2026-10-01 10:00:00.000 |                    6 |                                   157.44 |                    4 |
| +I | 2026-10-01 10:01:00.000 |                    6 |                                   247.44 |                    4 |
| +I | 2026-10-01 10:02:00.000 |                    6 |                                   337.44 |                    4 |

Guidelines

• Flink is stream-first; Spark is batch-first with streaming added. Choose Flink for sub-second latency.
• Use event time (not processing time) for accurate windowed aggregations.
• Watermarks handle late-arriving events — configure based on your latency tolerance.
• Windows that never emit usually mean a stalled watermark: a Kafka source with more subtasks than partitions does not go idle on its own. Add with_idleness(...) (DataStream) or table.exec.source.idle-timeout (SQL), or lower the parallelism.
• FlinkKafkaConsumer and FlinkKafkaProducer are gone from the Kafka connector for Flink 2.x; PyFlink still exports the Python wrappers but they fail with "Could not found the Java class". Use KafkaSource/KafkaSink with from_source and sink_to.
• Match the connector to the Flink minor version. Connector 5.0.0 is published for Flink 2.1 and 2.2 only, which is why Step 1 pins 2.2.1; before moving to a newer Flink, check that https://flink.apache.org/downloads/ lists a connector release for it.
• In SQL prefer the window functions TUMBLE, HOP, SESSION and CUMULATE in the FROM clause; the older GROUP BY TUMBLE(...) / TUMBLE_START form is deprecated.
• The json format reads TIMESTAMP as 2026-10-01 10:03:30.000. For 2026-10-01T10:03:30.000 add 'json.timestamp-format.standard' = 'ISO-8601' to the table options.
• Kafka sinks need checkpointing for AT_LEAST_ONCE and EXACTLY_ONCE; exactly-once also needs set_transactional_id_prefix(...) and consumers that read committed data only.
• Keep rest.bind-address: localhost as set in the cluster step: the web UI and REST API accept job submissions and have no login of their own, so reach a remote cluster through an SSH tunnel or a private network, never a public port.
• Flink is a cluster to operate (JobManager, TaskManagers, checkpoint storage). For small volumes or one-off batch work, a plain Kafka consumer or a batch engine is simpler.
• Managed Flink: Amazon Managed Service for Apache Flink (formerly Kinesis Data Analytics), Confluent Cloud, or Ververica Platform.