PyFlink Reference

Flink architecture, Flink SQL and the DataStream API in Python, event time and watermarks, state and checkpoints, connectors, deployment, and Flink vs Spark.

1. What Flink & PyFlink Are

Apache Flink is a distributed engine for stateful stream processing. It processes each event as it arrives (not in micro-batches), keeps large keyed state, handles event time with watermarks, and gives exactly-once state consistency through checkpoints. Typical latency is milliseconds. PyFlink is its Python API: you write jobs in Python, Flink runs them on the JVM cluster, and Python functions (UDFs, process functions) run in Python workers alongside it.

You can mix them: define sources in SQL, convert to a DataStream for custom logic, and convert back (t_env.to_data_stream / t_env.from_data_stream).

2. Architecture

 Client (python job.py / flink run -py)
        │  submits job graph
        ▼
 JobManager ── schedules tasks, coordinates checkpoints, recovers from failures
        │
   ┌────┴─────────────┬──────────────────┐
 TaskManager 1     TaskManager 2      TaskManager 3
 [slot][slot]      [slot][slot]       [slot][slot]     ← each slot runs one parallel slice
  source → map → keyBy → window → sink (operators chained into tasks)

• Parallelism is the number of parallel instances of each operator. Total slots must be at least the job's maximum parallelism. A Kafka source cannot usefully exceed the topic's partition count.
• keyBy repartitions the stream by key so all events for one key (e.g. an account) go to the same parallel instance and share its keyed state.
• Python code runs in a separate Python process per TaskManager by default (python.execution-mode: process). thread mode embeds Python in the JVM for lower overhead.

3. Setup

pip install apache-flink==1.20.0          # needs Java 11 (or 17) on the PATH

# connectors are JARs: download the one matching your Flink version, e.g.
# flink-sql-connector-kafka-3.3.0-1.20.jar

from pyflink.table import EnvironmentSettings, TableEnvironment

t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
t_env.get_config().set("pipeline.jars",
                       "file:///opt/jars/flink-sql-connector-kafka-3.3.0-1.20.jar")
t_env.get_config().set("parallelism.default", "2")

Running python job.py starts a local mini-cluster, which is ideal for development. The same file is submitted unchanged to a real cluster.

4. Flink SQL in PyFlink

A complete job: read JSON transactions from Kafka with an event-time watermark, and count transactions and amounts per channel every minute.

t_env.execute_sql("""
CREATE TABLE txn (
  txn_id     STRING,
  account_id STRING,
  channel    STRING,
  amount     DOUBLE,
  event_time TIMESTAMP(3),
  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'txn.events',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'flink-channel-agg',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
)""")

t_env.execute_sql("""
CREATE TABLE channel_minute (
  window_start TIMESTAMP(3), window_end TIMESTAMP(3),
  channel STRING, txns BIGINT, total DOUBLE
) WITH ('connector' = 'print')""")          # swap for kafka, jdbc, filesystem, iceberg …

t_env.execute_sql("""
INSERT INTO channel_minute
SELECT window_start, window_end, channel, COUNT(*), SUM(amount)
FROM TABLE(TUMBLE(TABLE txn, DESCRIPTOR(event_time), INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, channel
""").wait()                                  # .wait() keeps a local script running

5. DataStream API: Stateful Fraud Rule

A per-account velocity rule: alert when an account makes more than 5 transactions within 60 seconds of its first one. Keyed state holds each account's count, and a timer resets it.

import json
from pyflink.common import Types, WatermarkStrategy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment, KeyedProcessFunction, RuntimeContext
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer
from pyflink.datastream.state import ValueStateDescriptor

env = StreamExecutionEnvironment.get_execution_environment()
env.add_jars("file:///opt/jars/flink-sql-connector-kafka-3.3.0-1.20.jar")
env.enable_checkpointing(60_000)                       # exactly-once state, every minute

source = (KafkaSource.builder()
    .set_bootstrap_servers("kafka:9092")
    .set_topics("txn.events")
    .set_group_id("fraud-velocity")
    .set_starting_offsets(KafkaOffsetsInitializer.latest())
    .set_value_only_deserializer(SimpleStringSchema())
    .build())


class VelocityRule(KeyedProcessFunction):

    def open(self, runtime_context: RuntimeContext):
        self.count = runtime_context.get_state(ValueStateDescriptor("count", Types.INT()))

    def process_element(self, txn, ctx):
        n = (self.count.value() or 0) + 1
        self.count.update(n)
        if n == 1:                                      # first txn: reset after 60 s
            ts = ctx.timer_service().current_processing_time() + 60_000
            ctx.timer_service().register_processing_time_timer(ts)
        if n > 5:
            yield json.dumps({"account_id": txn["account_id"], "rule": "VELOCITY", "count": n})

    def on_timer(self, timestamp, ctx):
        self.count.clear()


(env.from_source(source, WatermarkStrategy.no_watermarks(), "kafka-txn")
    .map(json.loads)
    .key_by(lambda t: t["account_id"])
    .process(VelocityRule(), output_type=Types.STRING())
    .print())                                           # replace with a KafkaSink to fraud.alerts

env.execute("fraud-velocity-rule")

This rule uses processing time for simplicity. For event-time logic (e.g. "5 transactions within 60 seconds of transaction time"), assign a watermark strategy and use event-time timers.

6. Event Time & Watermarks

from pyflink.common import Duration, WatermarkStrategy
from pyflink.common.watermark_strategy import TimestampAssigner

class TxnTime(TimestampAssigner):
    def extract_timestamp(self, value, record_timestamp):
        return value["event_ms"]                       # epoch milliseconds

strategy = (WatermarkStrategy
    .for_bounded_out_of_orderness(Duration.of_seconds(5))   # tolerate 5 s of disorder
    .with_timestamp_assigner(TxnTime())
    .with_idleness(Duration.of_seconds(30)))                 # don't let idle partitions stall

• Watermark = highest event time seen − allowed out-of-orderness. A window fires when the watermark passes its end.
• Events later than the watermark are dropped from windows unless you configure allowed lateness or route them to a side output.
• The watermark of an operator is the minimum across its inputs, so one idle Kafka partition can stop all windows: always set idleness.
• The Streaming (Kafka & Kinesis) reference covers watermarks and their trade-offs in depth.

7. State, Checkpoints & Savepoints

from pyflink.datastream.state_backend import EmbeddedRocksDBStateBackend
from pyflink.datastream import CheckpointingMode

env.set_state_backend(EmbeddedRocksDBStateBackend())     # state on local disk; scales to large state
env.enable_checkpointing(60_000, CheckpointingMode.EXACTLY_ONCE)
env.get_checkpoint_config().set_checkpoint_storage_dir("s3://acme-flink/checkpoints/fraud")
env.get_checkpoint_config().set_min_pause_between_checkpoints(30_000)

# expire idle keys so state does not grow forever
from pyflink.common.time import Time
from pyflink.datastream.state import StateTtlConfig, ValueStateDescriptor

ttl = StateTtlConfig.new_builder(Time.days(1)).build()
desc = ValueStateDescriptor("last_seen", Types.LONG())
desc.enable_time_to_live(ttl)

Upgrade a job without losing state: flink stop --savepointPath s3://… <jobId>, deploy the new code, then flink run -s s3://…/savepoint-xyz -py job.py. Give operators stable UIDs so state can be mapped to the new job graph.

8. Connectors & Exactly-Once Sinks

End-to-end exactly-once needs three things: a replayable source (Kafka offsets in the checkpoint), checkpointed state, and a transactional or idempotent sink. With an at-least-once sink, make writes idempotent with upserts on a key.

9. Deploying PyFlink Jobs

# standalone / session cluster
./bin/flink run --python jobs/fraud_velocity.py \
    --jarfile /opt/jars/flink-sql-connector-kafka-3.3.0-1.20.jar

./bin/flink list                       # running jobs
./bin/flink cancel <jobId>

# Kubernetes: Flink Kubernetes Operator (application mode)
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: fraud-velocity
spec:
  image: myregistry/pyflink-jobs:1.4.0      # Flink + Python + job code + connector JARs
  flinkVersion: v1_20
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    state.checkpoints.dir: s3://acme-flink/checkpoints/fraud
  serviceAccount: flink
  jobManager:  { resource: { memory: "2048m", cpu: 1 } }
  taskManager: { resource: { memory: "4096m", cpu: 2 } }
  job:
    jarURI: local:///opt/flink/opt/flink-python-1.20.0.jar
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-pyclientexec", "/usr/local/bin/python3", "-py", "/opt/flink/usrlib/fraud_velocity.py"]
    parallelism: 4
    upgradeMode: savepoint

10. PyFlink vs Spark Structured Streaming

11. Common Interview Questions