text2flink

Alpha: runnable & tested, APIs may change.

Natural language → full, deployable, execution-verified Apache Flink jobs.

An open-source, agentic system that turns a plain-English request into a real Flink job and proves it works by running it against synthetic streaming data before handing it back.

New here? Start with docs/USE_CASES.md — who it's for, what it can express, and how to run it. See DESIGN.md for the architecture and roadmap.

It's a tool, not an assistant plugin. text2flink calls an LLM to draft a spec, then generates Flink SQL and verifies it by running on real Flink. You don't install it into Claude/Codex — you run it, import it, or let an assistant call it over MCP:

claude mcp add text2flink -- python3 -m text2flink.mcp_server

The assistant then calls generate_flink_job (also ground_kafka_topic, deploy_to_kafka) and gets back SQL that provably ran on Flink, not a guess. See docs/USE_CASES.md.

Status

Alpha — runnable and execution-verified today; APIs may change. The core loop is proven: generate Flink SQL from a structured IR, run it on a real local Flink cluster, assert streaming-specific properties (windows fire once, counts match a ground truth, event-time and watermarks respected) instead of brittle golden diffs, and repair a wrong job automatically from structured verification feedback.

What it does today

• NL → JobSpec extraction via a model-agnostic LLM layer that runs on OpenAI (OPENAI_API_KEY) or Anthropic (ANTHROPIC_API_KEY), with an offline heuristic proposer fallback so everything runs with no API key.
• Windowed aggregates — tumbling, hopping, and session windows (per-key inactivity-gap merging via the Flink SESSION TVF).
• Updating operators — unbounded GROUP BY (running totals), top-N, and deduplication (ROW_NUMBER()), verified via batch-mode collect and deployed to an upsert-kafka topic keyed by their partition keys.
• Joins — time-bounded interval joins of two event-time streams, and temporal (versioned-table) joins (FOR SYSTEM_TIME AS OF). Semantics confirmed against real Flink before being encoded in the oracle.
• Late-data / watermark correctness as a first-class, tested dimension: the oracle drops records whose event-time is at/behind the watermark, and a task verifies the generated job drops exactly those records — the thing batch Text2SQL cannot express.
• Live Kafka grounding — discover a source schema from a real Kafka topic (sample → infer → wire the connector), graded by running on real Flink against the real topic in bounded mode. Avro grounding is supported via the Flink avro format end-to-end; the Confluent Schema Registry (avro-confluent) path is verified offline against a mock registry.
• Deployable Kafka pipelines (topic → topic) — a spec compiles to a complete INSERT INTO <sink> SELECT … script; still verified by running to completion and grading the sink topic against the oracle.
• Multiple codegen targets — Flink SQL (execution-verified), a submittable PyFlink Table API program (flink run -py job.py), and Apache Spark SQL (a portability proof graded against the same oracle; the product stays Flink-first).

StreamBench

A declarative, contributable benchmark scored by execution pass rate, not string match — 22 tasks as JSON files in streambench/tasks/, grouped into core / advanced / adversarial tiers. Add a task or submit a model result without writing engine code (see CONTRIBUTING.md). Live results in LEADERBOARD.md:

Proposer	core	advanced	adversarial	total
openai-gpt-4o	6/6	9/9	7/7	22/22 (100%)
heuristic (offline)	6/6	9/9	0/7	15/22 (68%)

The adversarial tier (misleading phrasing, implicit/negated filters, multi-key grouping, reworded windows, non-second time units) makes the benchmark discriminating, and has surfaced real bugs no prior task exercised. Every task's gold spec is execution-verified on Flink (scripts/check_gold.py), so a broken task can't hide as a proposer failure.

Roadmap

Confluent Schema Registry (avro-confluent) end-to-end, DataStream codegen, K8s deployment manifests, and a bigger adversarial tier.

Requirements

This machine's default Java (26) and Python (3.14) are too new for Flink, so Phase 0 uses:

• Apache Flink + JDK 17, installed via Homebrew (brew install apache-flink openjdk@17).
• Python 3.11+ for the orchestrator (no PyFlink dependency — it drives Flink's SQL client as a subprocess, so the system Python is fine).
• For the live-Kafka example only: brew install kafka, plus the Flink Kafka SQL connector jar in Flink's lib/ (e.g. flink-sql-connector-kafka-5.0.0-2.2.jar from Maven Central).
• For the Avro example: the Flink Avro format jar in Flink's lib/ (e.g. flink-sql-avro-2.2.1.jar from Maven Central).
• For the (experimental) cross-engine example only: brew install apache-spark (JDK 17).

Run it

python3 examples/phase0_tumbling_count.py   # spike: verify a job, then catch & repair a broken one
python3 examples/phase1_streambench.py      # run StreamBench, report execution pass rate

phase1_streambench.py uses the offline heuristic proposer by default. To use a real model:

OPENAI_API_KEY=sk-... python3 examples/phase1_streambench.py       # OpenAI (default gpt-4o)
ANTHROPIC_API_KEY=sk-... python3 examples/phase1_streambench.py    # Anthropic

Layout

text2flink/
  ir.py          # IR: JobSpec (windows/agg) + IntervalJoinSpec + TemporalJoinSpec
  codegen.py     # IR -> Flink SQL: aggregates, interval joins, temporal joins
  data.py        # row-dict -> Flink CSV
  oracle.py      # reference interpreter: ground truth incl. late-drop + session windows
  runtime.py     # local Flink cluster lifecycle + run SQL, collect results
  assertions.py  # property-based streaming correctness checks
  verify.py      # candidate -> SQL -> run -> assert vs oracle  => VerifyResult
  llm.py         # model-agnostic LLM clients: OpenAI + Anthropic (raw HTTPS, no SDK)
  proposer.py    # NL -> JobSpec + repair: LLM / offline-heuristic / scripted proposers
  pipeline.py    # extract -> verify -> repair loop for one task
  kafka.py       # live Kafka grounding: sample a topic, infer schema, build a Source
  schema_registry.py # Avro/Schema-Registry grounding: exact types via avro-confluent
  deploy.py      # deployable INSERT jobs (append + upsert sinks), verify by consuming
  mcp_server.py  # MCP server (stdio) — lets an AI assistant call text2flink as a tool
  mcp_tools.py   # generate-and-run-on-Flink logic behind the MCP tool
  pyflink.py     # third codegen target: same IR -> a submittable PyFlink Table API program
  spark.py       # EXPERIMENTAL portability proof: same IR -> Spark SQL, same oracle
  streambench.py # loader for the StreamBench corpus
streambench/
  tasks/*.json   # the benchmark corpus (declarative, contributable)
examples/
  phase0_tumbling_count.py
  phase1_streambench.py
  phase2_kafka_grounding.py   # discover schema from a live Kafka topic, then verify
  phase2_kafka_pipeline.py    # deployable topic -> topic pipeline (INSERT INTO sink), verified
  phase3_avro_pipeline.py     # real Avro end-to-end on Flink (no Schema Registry)
  phase3_cross_engine.py      # (experimental) one IR compiled + verified on Flink AND Spark
tests/           # fast offline suite (no Flink): corpus, oracle, codegen, parsing
LICENSE          # Apache-2.0
CONTRIBUTING.md  # how to add a StreamBench task
.github/workflows/ci.yml   # runs the offline tests on 3.10–3.12

Development

pip install -e ".[dev]"
pytest                          # fast, offline — no Flink required
python3 scripts/check_gold.py   # verify every task's gold spec on Flink (task soundness)
python3 scripts/leaderboard.py  # score a proposer, update results/ + LEADERBOARD.md

Contributions welcome — the highest-value ones are a new StreamBench task (especially adversarial) and submitting a model to the leaderboard. See CONTRIBUTING.md.