flink-python

Follows python-skill conventions: PEP-585 types everywhere, Result[T] for fallible logic, structlog. Comments are for the person reading and reviewing the code: the public API, anything not obvious, and the reason behind the code as a plain sentence (no WHY: label). Keep the work plan out of comments: no phases, plan steps, ticket numbers, or words like for now / temporary / later — they go stale and someone has to find and delete them. Unfinished work belongs in plan.md. Any function that does real work (stateful operators, aggregates, caches, UDFs) gets a header comment above it: one line saying what it is for, plus its big-O for compute AND memory, derived DFS through the call tree — annotate per-record cost and state growth. The big-O lives in the header, not on a line inside the body. Tiny obvious helpers are exempt. Editing counts too: when you change a function, keep the comments already there, fix any your change made wrong, and add the missing header / big-O / reason. Do this only for code you add or change, not for nearby code you did not touch.

When to Use PyFlink

Rule: PyFlink adds ~10-30% overhead vs Java in the default process mode (see Execution Mode under Performance). Use for UDF-heavy workloads, not raw throughput.

Environment Setup

Targets apache-flink 2.x (pip install apache-flink), Python 3.9-3.12. Flink 1.x-era APIs (register_function, legacy SourceFunction) are gone — do not generate them.

Table Environment (Recommended)

from pyflink.table import EnvironmentSettings, TableEnvironment

# streaming mode
env_settings: EnvironmentSettings = EnvironmentSettings.in_streaming_mode()
table_env: TableEnvironment = TableEnvironment.create(env_settings)

# batch mode
env_settings_batch: EnvironmentSettings = EnvironmentSettings.in_batch_mode()
table_env_batch: TableEnvironment = TableEnvironment.create(env_settings_batch)

# configuration
table_env.get_config().set("table.exec.state.ttl", "24 h")
table_env.get_config().set("parallelism.default", "4")

DataStream Environment

from pyflink.datastream import StreamExecutionEnvironment, CheckpointingMode
from pyflink.table import StreamTableEnvironment

env: StreamExecutionEnvironment = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)

# checkpointing
env.enable_checkpointing(60000)
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)

table_env: StreamTableEnvironment = StreamTableEnvironment.create(env)

Dependencies

from pyflink.table import TableEnvironment

table_env: TableEnvironment = ...

# third-party packages: installed into the UDF workers' environment
table_env.set_python_requirements("/path/to/requirements.txt")

# your own modules: shipped to workers and put on PYTHONPATH (never installed)
table_env.get_config().set("python.files", "/path/to/my_package")

# jar dependencies (connectors)
jar_paths: str = (
    "file:///path/to/flink-sql-connector-kafka.jar;"
    "file:///path/to/flink-sql-avro-confluent-registry.jar"
)
table_env.get_config().set("pipeline.jars", jar_paths)

Rule: python.files is PYTHONPATH-only — it installs nothing. Third-party requirements go through set_python_requirements() (config key python.requirements).

Result[T] at UDF Boundaries

PyFlink UDFs must return Flink DataTypes (STRING, DOUBLE, etc.) — they can't return Result[T] directly. Use Result for internal validation/parsing logic, then unwrap at the eval() boundary.

import json
import structlog

log: structlog.stdlib.BoundLogger = structlog.get_logger()

# Internal logic uses Result
def validate_payload(raw: str) -> Result[dict[str, object]]:
    result: Result[dict[str, object]] = try_call(lambda: json.loads(raw), "PARSE")
    match result:
        case Err():
            return result
        case Ok(data):
            pass
    if "user_id" not in data:
        return err("VALIDATION", "missing user_id")
    if "event_type" not in data:
        return err("VALIDATION", "missing event_type")
    return ok(data)

# At UDF boundary: unwrap Ok, return None for Err
@udf(result_type=DataTypes.STRING())
def extract_user_id(raw: str | None) -> str | None:
    if raw is None:
        return None
    match validate_payload(raw):
        case Ok(data):
            return str(data["user_id"])
        case Err(failure):
            log.warning("validation_failed", error=str(failure))
            return None  # UDFs can't return Result — None signals failure

Rule: Result inside, unwrap at the Flink boundary. Log failures — don't silently return None.

Scalar UDFs

Decorator Style (Row → Value)

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())
def normalize_email(email: str | None) -> str | None:
    if email is None:
        return None
    normalized: str = email.lower().strip()
    return normalized

table_env.create_temporary_function("normalize_email", normalize_email)

table_env.execute_sql("""
    select
      user_id,
      normalize_email(email) as email
    from users
""")

Class-Based with Resources (open / close)

import re
from re import Pattern

from pyflink.table import DataTypes
from pyflink.table.udf import udf, ScalarFunction, FunctionContext

class TokenizerUDF(ScalarFunction):
    def __init__(self) -> None:
        self.pattern: Pattern[str] | None = None

    def open(self, function_context: FunctionContext) -> None:
        self.pattern = re.compile(r"\W+")

    def eval(self, text: str | None) -> str | None:
        assert self.pattern is not None  # wired in open() — fail loudly, never drop rows
        if text is None:
            return None
        tokens: list[str] = [t for t in self.pattern.split(text.lower()) if t]
        return ",".join(tokens)

    def close(self) -> None:
        self.pattern = None

tokenizer_udf: ScalarFunction = TokenizerUDF()
tokenizer = udf(tokenizer_udf, result_type=DataTypes.STRING())
table_env.create_temporary_function("tokenize", tokenizer)

Rules: Load heavy resources in open(), release in close(). Never load in eval(). State wired in open() is an invariant — assert it in eval(); keep None-checks only for genuinely nullable Flink inputs.

Table UDF (Row → Rows)

from collections.abc import Iterator

from pyflink.table import DataTypes
from pyflink.table.udf import udtf

@udtf(result_types=[DataTypes.STRING(), DataTypes.INT()])
def split_tags(tags: str | None) -> Iterator[tuple[str, int]]:
    if tags is None:
        return
    for tag in tags.split(","):
        stripped: str = tag.strip()
        if stripped:
            yield stripped, len(stripped)

table_env.create_temporary_function("split_tags", split_tags)

table_env.execute_sql("""
    select
      event_id,
      tag,
      tag_length
    from events
    cross join lateral table(split_tags(tags)) as t(tag, tag_length)
""")

Aggregate UDF (Rows → Value)

from dataclasses import dataclass

from pyflink.table import DataTypes
from pyflink.table.types import DataType
from pyflink.table.udf import udaf, AggregateFunction

@dataclass
class WeightedAvgAcc:
    total: float = 0.0
    weight_sum: float = 0.0

class WeightedAvg(AggregateFunction):
    def get_value(self, acc: WeightedAvgAcc) -> float | None:
        if acc.weight_sum == 0:
            return None
        return acc.total / acc.weight_sum

    def create_accumulator(self) -> WeightedAvgAcc:
        return WeightedAvgAcc()

    def accumulate(self, acc: WeightedAvgAcc, value: float | None, weight: float | None) -> None:
        if value is not None and weight is not None:
            acc.total += value * weight
            acc.weight_sum += weight

    def retract(self, acc: WeightedAvgAcc, value: float | None, weight: float | None) -> None:
        if value is not None and weight is not None:
            acc.total -= value * weight
            acc.weight_sum -= weight

    def get_accumulator_type(self) -> DataType:
        return DataTypes.ROW([
            DataTypes.FIELD("total", DataTypes.DOUBLE()),
            DataTypes.FIELD("weight_sum", DataTypes.DOUBLE()),
        ])

    def get_result_type(self) -> DataType:
        return DataTypes.DOUBLE()

weighted_avg = udaf(WeightedAvg())
table_env.create_temporary_function("weighted_avg", weighted_avg)

Pandas UDFs (Vectorized)

PyFlink pins pandas >=1.3,<2.3 inside workers — write pandas-1.x/2.x-compatible code here. Do not apply pandas 3.x idioms (python-numerics-skill's pandas 3.0 guidance does not apply inside Flink workers).

Scalar Pandas UDF

import pandas as pd

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.DOUBLE(), func_type="pandas")
def pandas_zscore(values: pd.Series) -> pd.Series:
    mean: float = values.mean()
    std: float = values.std()
    if std == 0:
        return pd.Series([0.0] * len(values))
    return (values - mean) / std

table_env.create_temporary_function("zscore", pandas_zscore)

Aggregate Pandas UDF

import pandas as pd

from pyflink.table import DataTypes
from pyflink.table.udf import udaf

@udaf(result_type=DataTypes.DOUBLE(), func_type="pandas")
def pandas_median(values: pd.Series) -> float:
    return values.median()

table_env.create_temporary_function("median", pandas_median)

Batch Pandas UDF (Multiple Columns → ROW)

The only way to return several columns from one vectorized pass: result_type=DataTypes.ROW([...]) with a pd.DataFrame return. Plain-function pandas UDFs have no open() — lazy-load shared resources at first call (full model-lifecycle patterns: references/ml-inference.md).

from pathlib import Path
from typing import Any

import joblib
import numpy as np
import pandas as pd
from numpy.typing import NDArray

from pyflink.table import DataTypes
from pyflink.table.udf import udf

class LazyModel:
    def __init__(self, model_path: str) -> None:
        self.model_path: str = model_path
        self._model: Any = None

    def get(self) -> Any:
        if self._model is None:  # loads once per Python worker, not per batch
            # joblib/pickle deserialization executes code: load only model
            # artifacts you own (your registry/CI), never user-supplied paths
            self._model = joblib.load(Path(self.model_path))
        return self._model

classifier: LazyModel = LazyModel("/models/classifier.joblib")

@udf(
    result_type=DataTypes.ROW([
        DataTypes.FIELD("prediction", DataTypes.DOUBLE()),
        DataTypes.FIELD("confidence", DataTypes.DOUBLE()),
    ]),
    func_type="pandas",
)
def predict_batch(
    feature1: pd.Series,
    feature2: pd.Series,
    feature3: pd.Series,
) -> pd.DataFrame:
    features: pd.DataFrame = pd.DataFrame({
        "f1": feature1, "f2": feature2, "f3": feature3,
    })
    predictions: NDArray[np.float64] = classifier.get().predict_proba(features)
    return pd.DataFrame({
        "prediction": predictions[:, 1],
        "confidence": predictions.max(axis=1),
    })

ML Model Inference

WHEN running sklearn/torch/joblib models inside UDFs → read references/ml-inference.md. Covers: sklearn predictor with Result-wrapped prediction, TorchScript inference (cpu, no_grad), batch inference via pandas UDF with a lazy-loading predictor class. Rules that always apply: models load in open() (or lazily for plain-function pandas UDFs), eval() handles None inputs, prediction failures are logged and mapped to None.

DataStream API

WHEN writing DataStream jobs — Kafka sources, watermarks, DLQ side outputs, keyed state, timers → read references/datastream.md. Covers: KafkaSource builder + watermark strategy with idleness, parse-with-Result routing errors to a DLQ via OutputTag, ValueState/ListState, event-time timers for session detection (including the PyFlink generator-protocol trick). Rules that always apply: watermarks assigned after parsing, every operator gets a .uid(), parse/validation errors route to a DLQ side output — never dropped.

Performance Optimization

Execution Mode (process vs thread)

python.execution-mode (default process) is the single biggest PyFlink performance lever. thread runs UDFs inside the JVM via PEMJA, eliminating the IPC and serialization overhead between the operator and a separate Python worker process — exactly the ~10-30% overhead noted above.

table_env.get_config().set("python.execution-mode", "thread")

Thread-mode limits (Flink 2.2): general Python scalar UDFs and UDTFs only — Python UDAFs and all pandas UDFs are unsupported and fall back to process mode; UDFs share the JVM process and the GIL still serializes Python execution. Prefer thread mode for high-QPS scalar-UDF pipelines; keep process mode for isolation or pandas/UDAF workloads.

Bundle and Arrow Batch Sizing

table_env: TableEnvironment = ...

# defaults are 1000 / 1000ms — raise size for throughput, lower time for latency
table_env.get_config().set("python.fn-execution.bundle.size", "5000")
table_env.get_config().set("python.fn-execution.bundle.time", "1000")  # ms

# arrow batch (pandas UDFs) is capped by bundle.size — keep it <= bundle.size
table_env.get_config().set("python.fn-execution.arrow.batch.size", "5000")

Memory

python.fn-execution.memory.managed defaults to true: Python workers draw from the task manager's managed memory and need no extra configuration. Only if you set it to false must you size taskmanager.memory.task.off-heap.size for the Python processes.

Caching Heavy Resources

from typing import Any, ClassVar

from pyflink.table.udf import ScalarFunction, FunctionContext

class HeavyUDF(ScalarFunction):
    _shared_resource: ClassVar[Any] = None  # class-level cache: one load per process, not per slot

    def open(self, function_context: FunctionContext) -> None:
        if HeavyUDF._shared_resource is None:
            HeavyUDF._shared_resource = load_heavy_model()
        self.resource: Any = HeavyUDF._shared_resource

    def eval(self, value: str | None) -> str | None:
        if value is None:
            return None
        return self.resource.process(value)

Anti-Patterns

Rules (full broken/fixed pairs: references/anti-patterns.md):

• Load heavy resources in open(), never in eval() — eval() runs per row.
• Lambdas pickle fine (PyFlink serializes UDFs with cloudpickle). What breaks is a non-picklable resource (DB connection, file handle, lock, loaded model) held as an instance attribute at submission time — construct those in open().
• Never swallow errors silently — log with structlog before mapping to None.
• SQL NULL arrives as Python None — every UDF parameter is | None and must be handled.
• Python-side caches must be bounded (lru_cache) or use Flink state — unbounded dicts grow forever per worker.

Code Review Checklist

• Type annotations: Every variable has explicit PEP-585 type (name: str = ...)
• Union syntax: str | None not Optional[str]
• Result[T]: Internal validation uses Result, unwrapped at eval() boundary
• Failures logged: Err paths log with structlog before returning None
• Resource loading: Heavy resources in open(), released in close()
• Asserts: State/resources wired in open() asserted non-None in eval()/process_element() — wiring bugs fail loudly, never drop records
• Null handling: All UDFs handle None inputs explicitly (SQL NULL arrives as Python None)
• Serializable: Non-picklable resources constructed in open(), never held as attributes at submission (lambdas are fine — cloudpickle)
• Complexity: Stateful operators and aggregates annotated with big-O for per-record compute AND state memory
• Execution mode: python.execution-mode chosen deliberately — thread for scalar-UDF throughput, process for pandas/UDAF/isolation
• Bundle size: Configured for throughput vs latency tradeoff; arrow batch <= bundle size
• Memory: Managed memory (default) understood; off-heap sized only when python.fn-execution.memory.managed=false
• State: Python-side caches bounded (lru_cache) or use Flink state
• DLQ: DataStream pipelines route parse/validation errors via side outputs
• Watermarks: Assigned after parsing, with idleness configured
• Dependencies: set_python_requirements() for packages, python.files for own modules, pipeline.jars for connectors
• Testing: UDFs unit tested outside Flink first — TDD: red, green, vacuous mutation check; eval() is a plain method, test it as one