Streamform — Flink backend

The Flink backend runs the same Streaming IR the simulator runs, on Apache Flink. It generates Flink SQL; it does not generate Java or Python, and Streamform never links a Flink client. streamform build --backend flink writes the script and streamform apply submits it through Flink’s SQL Gateway.

What streamform build --backend flink generates

For examples/customer-metrics with the local target:

CREATE TABLE `orders` (
  `amount` DECIMAL(38, 9),
  `customer_id` BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json',
  'scan.startup.mode' = 'earliest-offset'
);

CREATE TEMPORARY VIEW `customer_metrics` AS
SELECT `customer_id`, `sum(amount)` AS `total_spend` FROM (SELECT `customer_id`, SUM(`amount`) AS `sum(amount)` FROM `orders` GROUP BY `customer_id`) AS t3;

CREATE TABLE `sink_customer_metrics` (
  `customer_id` BIGINT,
  `total_spend` DECIMAL(38, 9),
  PRIMARY KEY (`customer_id`) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'customer-metrics',
  'properties.bootstrap.servers' = 'kafka:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.json.encode.decimal-as-plain-number' = 'true',
  'value.json.encode.decimal-as-plain-number' = 'true'
);

EXECUTE STATEMENT SET
BEGIN
  INSERT INTO `sink_customer_metrics` SELECT * FROM `customer_metrics`;
END;

The rules, each derived from the plan rather than from anything Flink-specific in it:

• Every source is a CREATE TABLE. A kafka source becomes the kafka connector with the target’s bootstrap servers, the source’s topic and format, and scan.startup.mode from the target’s kafka.startup. A file source becomes the filesystem connector.

• Every model is a CREATE TEMPORARY VIEW, built from the model’s operator chain as nested subqueries. ref('m') reads the view m, so a chain of models runs inside one job without a topic in between.

• A model with a sink: block also gets a table named sink_<model> and an INSERT INTO inside the single EXECUTE STATEMENT SET. materialized: append uses the kafka connector; materialized: upsert uses upsert-kafka with PRIMARY KEY (…) NOT ENFORCED on the model’s key, so a delete in the changelog is a tombstone on the topic. A file sink cannot carry an upsert changelog and is refused.

• Types map from the source vocabulary: int32 → INT, int64 → BIGINT, float64 → DOUBLE, decimal(p,s) → DECIMAL(p, s), string → STRING, boolean → BOOLEAN, timestamp → TIMESTAMP(3).

• Identifiers are backtick-quoted, including the names Streamform gives unaliased aggregate columns (sum(amount)); the model’s own column names are what a reader sees.

• Event time (event time): a source with event_time gets WATERMARK FOR <column> AS <column> - INTERVAL '<delay>' (rendered in the largest whole unit: '5' MINUTE, '1' HOUR, '1.500' SECOND) and, for JSON, 'json.timestamp-format.standard' = 'ISO-8601'. With late_events: drop the table is created as <source>_raw and the name models read becomes a CREATE TEMPORARY VIEW keeping only rows with CURRENT_WATERMARK(<column>) IS NULL OR <column> >= CURRENT_WATERMARK(<column>). A model’s retention becomes a STATE_TTL hint on its aggregate, which Flink enforces in processing time; build and apply warn that it is approximate and wrong under replay. Flink’s ISO-8601 for TIMESTAMP(3) takes no zone designator: producers write 2026-08-30T10:00:00, not …Z.

A model with neither a sink nor a reader is planned but nothing runs it; build warns.

Output goes to build/flink/application.sql under the project, or to standard output with --stdout. Exit code 1 means the backend cannot represent the application (a capability error is printed); 2 means the project itself does not pass streamform check.

streamform apply

streamform apply [--target NAME] builds the application and submits each statement, in order, to the target’s Flink SQL Gateway over its REST API: it opens a session, runs every CREATE TABLE and CREATE TEMPORARY VIEW, runs the EXECUTE STATEMENT SET, prints the job id Flink assigns, and closes the session.

applying customer-metrics to target `local` (http://localhost:8083)

ok: source orders
ok: model customer_metrics
ok: sink customer_metrics
ok: statement set → job 9d1f2c4b7a3e5f6081c2d3e4f5a6b7c8

ok: applied

If Flink rejects a statement, apply prints the statement it was on and Flink’s reason (the message lines of its exception, without the stack trace), stops, and exits 1; no job is started. A gateway that cannot be reached is exit 2. The gateway address comes from the target’s flink.gateway; it is plain HTTP, and the Flink cluster must have the Kafka SQL connector jar on its classpath for the generated tables to resolve.

Kafka topics a source reads must exist before apply: Flink’s Kafka source lists a topic’s partitions when the job starts and fails if there are none. Create them first (kafka-topics.sh --create) rather than relying on the first produced event to create them.

One Flink characteristic to know when consuming an upsert topic: a filter over an aggregate (a WHERE in a model that reads an upsert model, or HAVING) runs in Flink’s retract mode, and the upsert-kafka sink writes each retraction as a tombstone. An update to a row that stays inside the filter therefore reaches the topic as a tombstone followed by the new value for the same key, where the simulator’s changelog has one update. The keyed state a consumer holds is identical once the pair has arrived; a consumer that reacts to every record will see the key vanish and reappear.

apply submits and reports; it does not remember the job. Stopping, upgrading, and reconciling a running application are not built yet.

The integration environment

integration/docker-compose.yml runs one Kafka broker (apache/kafka:3.9.2, KRaft) and a Flink 2.0.2 cluster (JobManager, TaskManager, SQL Gateway) built from integration/flink/Dockerfile, which adds flink-sql-connector-kafka 4.0.1-2.0 to Flink’s classpath. Inside that network Kafka is kafka:9092, which is what the examples’ local target names; the host reaches the SQL Gateway at http://localhost:8083, the Flink UI at http://localhost:8081, and Kafka at localhost:9094. A release publishes both files in streamform-examples.tar.gz as flink-local/, next to the example projects, so the commands below work outside this repository with flink-local/docker-compose.yml and --project customer-metrics.

docker compose -f integration/docker-compose.yml up -d --build --wait
docker exec streamform-kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders
streamform apply --project examples/customer-metrics
printf '%s\n' '{"customer_id": 42, "amount": 10}' '{"customer_id": 42, "amount": 18}' \
  | docker exec -i streamform-kafka /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders
docker exec streamform-kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic customer-metrics --from-beginning --property print.key=true
docker compose -f integration/docker-compose.yml down -v

mise run integration does that around the differential test (crates/streamform-cli/tests/integration.rs): for each example fixture whose model declares a sink, it applies the project, produces the fixture’s given events to the source topic, reads the sink topic, and compares the (key, value | tombstone) sequence with what the simulator’s changelog implies, reading Flink’s tombstone-then-value pair for one key as the single update it is. Both backends running the same IR must agree; big_customers proves the delete arrives as a tombstone. The tests are #[ignore], so mise run check never needs Docker; .github/workflows/integration.yml runs them on demand and weekly.