PyFlink Cheat Sheet

2TableEnvironment & DDLconnect to sources

from pyflink.table import EnvironmentSettings, TableEnvironment t_env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())★
The entry point. in_batch_mode() for bounded/batch — the same API, unified.
t_env.execute_sql(""" CREATE TABLE orders (id BIGINT, amount DOUBLE, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND) WITH ('connector'='kafka', 'topic'='orders', 'format'='json', ...)""")★
Define sources/sinks as tables via SQL DDL. The WITH clause picks a connector (kafka, filesystem, jdbc); WATERMARK enables event-time.
t_env.execute_sql("CREATE TABLE sink (...) WITH ('connector'='print')")
The print connector is great for local debugging.

3Query & sinktransform & write

result = t_env.sql_query("SELECT id, SUM(amount) FROM orders GROUP BY id")★
Full streaming SQL — joins, aggregations, windows. Returns a Table.
t = t_env.from_path("orders") t.filter(col("amount") > 100).group_by(col("id")).select(...)
Or the fluent Table API (select/filter/group_by/join) instead of SQL strings.
result.execute_insert("sink")★
Writes to a sink AND triggers the job. For a streaming query this runs continuously. t_env.execute_sql("INSERT INTO sink SELECT ...") is the pure-SQL equivalent.
t_env.sql_query("SELECT ...").execute().print() # inspect results
Print a bounded result to the console for debugging.

4Time, watermarks, windowsthe streaming core

WATERMARK FOR ts AS ts - INTERVAL '5' SECOND★
Event-time processing: watermarks tell Flink "no events older than this are coming", allowing bounded lateness before closing windows. Defined in the table DDL.
# Tumbling window aggregation in SQL: SELECT window_start, SUM(amount) FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(ts), INTERVAL '1' MINUTE)) GROUP BY window_start, window_end★
Windowing TVFs: TUMBLE (fixed), HOP (sliding), CUMULATE, SESSION — the standard way to aggregate over time.
processing-time vs event-time
Event-time (with watermarks) gives correct, replayable results despite out-of-order data; processing-time is simpler but non-deterministic.

5UDFscustom Python logic

from pyflink.table.udf import udf @udf(result_type="BIGINT") def add_one(x): return x + 1★
Scalar UDF. Register (t_env.create_temporary_function) or use inline. Also @udtf (table), @udaf (aggregate).
@udf(result_type=..., func_type="pandas")★
Vectorized (pandas) UDFs process batches as pandas Series — far faster than row-at-a-time Python.

6Environment & transformsexplicit dataflow

from pyflink.datastream import StreamExecutionEnvironment env = StreamExecutionEnvironment.get_execution_environment()★
The DataStream entry point. Add sources (from_collection, Kafka connector) to get a DataStream.
ds.map(lambda x: (x[0], 1)) · ds.filter(...) · ds.flat_map(...)★
Stateless per-element transforms — like an RDD/Stream. Provide output_type= (Types) for correct serialization.
ds.key_by(lambda x: x[0]).reduce(lambda a, b: (a[0], a[1]+b[1]))★
key_by partitions the stream by key — required before any keyed operation (reduce, window, keyed state). Like a streaming GROUP BY.
env.execute("my-job")★
Submits & runs the DataStream job (the required final call).

7Keyed state & process fnsstateful streaming

class Counter(KeyedProcessFunction): def open(self, ctx): self.count = ctx.get_state(ValueStateDescriptor("c", Types.LONG()))★
Keyed state (per-key, fault-tolerant): ValueState, ListState, MapState. Declared in open(), scoped to the current key.
def process_element(self, value, ctx): self.count.update((self.count.value() or 0) + 1)
Read/update state per event. ctx.timer_service().register_event_time_timer(...) for timers (timeouts, windows).
ds.key_by(...).process(Counter())
Attach the process function to a keyed stream. This is the general-purpose stateful primitive.
env.enable_checkpointing(60000) # exactly-once fault tolerance★
Checkpointing snapshots state periodically for exactly-once recovery — essential for production stateful jobs.

8Interop & submitrun it

table = t_env.from_data_stream(ds) ds2 = t_env.to_data_stream(table)★
Convert between the two APIs — do the heavy relational work in SQL, drop to DataStream for the custom bit.
flink run -py job.py · flink run -pym module -pyfs deps/★
Submit to a real cluster with the flink CLI; ship Python deps/files with -pyfs/-pyreq.
t_env.get_config().set("parallelism.default", "4")
Configure parallelism, state backend, checkpoint dir via the environment config.

!Common gotchasread before shipping

nothing runs without execute()
The dataflow is lazy. Table jobs run on execute_insert/execute_sql(INSERT); DataStream on env.execute().
watermarks or windows never fire
Event-time windows only emit once watermarks advance past them. Idle sources / a too-large delay = windows that seem stuck. Define watermarks correctly.
key_by before keyed ops
Windows, reduce, and keyed state require a key_by first — and the key must be deterministic & hashable.
specify types
Python lambdas lose type info; pass output_type=Types.XXX() or serialization defaults to slow pickling.
checkpointing for stateful prod
Without enable_checkpointing, a failure loses all state. Turn it on (with a durable state backend) for anything real.

PyFlink cheat sheet