Running on Apache Flink

The Flink backend runs the same plan the simulator runs. It generates Flink SQL, which you can read and diff, and submits it through Flink’s SQL Gateway. Streamform never links a Flink client and generates no Java or Python.

Declare a target

A target is a place the application is built for and applied to: a SQL Gateway and the Kafka it reads and writes.

# streamform.yml
targets:
  local:
    backend: flink
    flink:
      gateway: http://localhost:8083
    kafka:
      bootstrap_servers: kafka:9092
      startup: earliest

bootstrap_servers is as Flink sees it (inside the bundled Compose network that is kafka:9092); gateway is as your machine sees it. build and apply take --target NAME, defaulting when the project declares exactly one.

Build

streamform build --backend flink            # writes build/flink/application.sql and build/plan.json
streamform build --backend flink --stdout   # prints the SQL, writes nothing

For examples/customer-metrics:

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 are few and each follows from the plan: every source is a CREATE TABLE; every model is a CREATE TEMPORARY VIEW built from its operator chain; a model with a sink: block also gets a sink_<model> table and an INSERT INTO in the single statement set. An upsert model uses upsert-kafka with a primary key on its declared key, so a delete in the changelog is a tombstone on the topic. A file sink cannot carry an upsert changelog and is refused with exit code 1. The Flink backend reference lists the type mapping and every rule.

Apply

streamform apply --target local

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

apply builds, opens a gateway session, runs each statement in order, prints the job id, and closes the session. If Flink rejects a statement, apply prints the statement it was on and Flink’s reason without the stack trace, stops with exit 1, and no job is started. An unreachable gateway is exit 2.

apply submits and reports; it does not remember the job. Stopping, upgrading, and reconciling a running application are not built yet; Preparing for upgrades covers what exists today toward that.

A local cluster in one command

The examples archive ships flink-local/docker-compose.yml: one Kafka broker (KRaft) and a Flink 2.0.2 cluster (JobManager, TaskManager, SQL Gateway) with the Kafka SQL connector on its classpath. From the streamform-examples directory:

docker compose -f flink-local/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 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 flink-local/docker-compose.yml down -v

The consumer prints two records for key {"customer_id":42} with totals 10 and 28, written at the column’s scale as 10.000000000 and 28.000000000: the simulator’s changelog, on Kafka. JSON sinks always write a decimal as a plain number ('json.encode.decimal-as-plain-number'); left to its default, Flink would write 10 as 1E+1. The Flink UI is at http://localhost:8081.

Two things the real cluster teaches

Create source topics before apply. Flink’s Kafka source lists a topic’s partitions at startup and fails if there are none; do not rely on the first produced event to create the topic.
A filter over an aggregate runs in Flink’s retract mode. An update to a row that stays inside the filter reaches an upsert-kafka 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 both have arrived; a consumer that reacts to every record will see the key vanish and reappear.

The differential test

In the source repository, mise run integration brings the environment up and runs the differential test: 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. Both backends run the same plan, so they must agree, and big_customers proves the delete arrives as a tombstone. The tests are #[ignore] so the ordinary check never needs Docker; CI runs them on demand and weekly.

Keyboard shortcuts