Lab: Migration using AI¶

The current AI based migration implementation supported by this tool enables migration of:

• Spark SQL to Flink SQL
• ksqlDB to Flink SQL

The approach uses LLM agents local or remote. After this lab you should be able to use the migration-to-flink tools to partially automate your SQL migration to Flink SQL.

The core idea is to leverage LLMs and parser tools to understand the source SQL semantics and to translate them to Flink SQLs.

This is github repository is not production ready, the LLM can generate hallucinations, and one to one mapping between source like ksqlDB or Spark to Flink is sometime not the best approach. We expect that this agentic solution could be a strong foundation for better results, and can be enhanced over time.

Migration is a one time shot, and should not be a practice to develop Flink solution.

Prerequisites¶

Be sure to have done the Setup Lab and Setup script to get different CLIs operational and generate Cursor/Claude skill variants from the canonical Agno skill/ directories.

Different runtimes¶

Agno harness (CLI) — translation and validation run via Python agents and ksql-flink-migrate / spark-flink-migrate. The harness loads skill/SKILL.md directly. Flink SQL validation uses flink-skill-validate or skill scripts under flink-skill-common/skill/scripts/.

Cursor (IDE) — skills under .cursor/skills/ are generated with MCP-oriented instructions. You (the IDE assistant) translate SQL using the skill rules. Validation and deploy use the flink-skill-common MCP server (validate_flink_sql_offline, create_flink_statement, etc.). Do not run ksql-flink-migrate or spark-flink-migrate in the IDE workflow. Enable MCP in Cursor Settings.

Claude Code (IDE) — skills under */.claude/skills/. You translate SQL using the skill rules. Validation uses flink-skill-validate CLI or bundled validate_offline.py scripts. Deploy uses the flink-skill-common MCP server when configured (flink-skill-mcp). Do not run Agno migration CLIs for IDE translation.

After editing any canonical skill/SKILL.md, run ./scripts/adapt-skills.sh --target cursor and/or ./scripts/adapt-skills.sh --target claude before using IDE workflows.

Claude Code integration¶

This section covers two skill scopes: Flink SQL validation and ksqlDB to Flink migration. Skills provide agent playbooks; CLIs run deterministic checks and translation.

Setup¶

Assume Setup Lab has already run (./scripts/setup.sh). That generates Claude skill copies under each package's .claude/skills/.

1. Environment — repo-root .env with optional FLINK_* for remote validation and deploy.
2. Optional remote validation / deploy — fill FLINK_* in .env when using flink-skill-validate remote or MCP deploy tools (create_flink_statement, etc.).

Scope 1: Flink SQL validation¶

Happy path¶

Use when you already have Flink DDL and/or DML and need syntax checks or convention fixes, and deployment using Confluent Cloud.

CLI (from repo root):¶

cd flink-skill-common/harness && uv sync --extra dev

uv run flink-skill-validate offline \
  --ddl ../../references/flink/valid/raw_classical_songs/ddl.raw_classical_songs.sql \
  --dml ../../references/flink/valid/raw_classical_songs/dml.raw_classical_songs.sql

The CLI prints JSON with "ok": true/false and an issues array. Exit code 0 means pass; 1 means validation errors.

Claude Code¶

Start claude code under the references/flink folder.

Load the validate-flink-sql skill. Run flink-skill-validate offline. on the Flink SQL in valid/filtering/ . If validation fails, apply the skill rules, write corrected SQL, and re-run until ok is true.

You should get a response like:

Validation passed on the first run — no fixes needed.

Result:
{
  "ok": true,
  "issues": [],
  "error_count": 0
}

I loaded the validate-flink-sql skill and ran flink-skill-validate offline on the two Flink SQL files in valid/filtering/:

- ddl.filtered_pub.sql — CREATE TABLE filtered_publications (has PRIMARY KEY ... NOT ENFORCED, DISTRIBUTED BY HASH(book_id), and standardized connector properties)
- dml.filtered_pub.sql — INSERT INTO filtered_publications ... SELECT ... WHERE author = 'George R. R. Martin'

Both are syntactically valid per the offline (sqlglot Flink-dialect) check: ok is true with zero issues. No corrections were required, so no re-run was needed. These files live under valid/, so passing is the expected outcome.

Invalid DML or DDL¶

Validate broken DML from references/flink/invalid/dml_bad_syntax/ (contains an INSRT INTO typo):

CLI¶

cd flink-skill-common/harness
uv run flink-skill-validate offline --dml ../../references/flink/invalid/dml_bad_syntax/dml.sql

Expected: exit code 1 and a json like:

{
  "ok": false,
  "issues": [
    {
      "statement_index": 0,
      "kind": "dml",
      "message": "DML must start with INSERT INTO",
      "line": null,
      "severity": "error"
    }
  ],
  "error_count": 1
}

Claude Code¶

Follow-up: Ask Claude to fix the DML using validate-flink-sql rules and re-run validation until it passes.

using /validate-flink-sq assess invalid/dml_bad_syntax/dml.sql

you should get something like:

Validation failed — as expected for this invalid/ fixture.

File: invalid/dml_bad_syntax/dml.sql
INSRT INTO raw_classical_songs(artist, title) VALUES ('Test Artist', 'Test Title');

Result:
{
  "ok": false,
  "issues": [
    {
      "statement_index": 0,
      "kind": "dml",
      "message": "DML must start with INSERT INTO",
      "line": null,
      "severity": "error"
    }
  ],
  "error_count": 1
}

Assessment: The statement's keyword is misspelled — INSRT instead of INSERT — so the validator rejects it as not starting with INSERT INTO. The corrected form would be:

INSERT INTO raw_classical_songs (artist, title) VALUES ('Test Artist', 'Test Title');

Since this lives under invalid/ (a negative test fixture meant to fail), I've left it unchanged and only reported the finding. Let me know if you'd like me to write the corrected version somewhere.

Deploy failure fix loop (IDE)¶

Use when Flink SQL is deployed to Confluent Cloud and a statement fails or is unhealthy. The host assistant performs the fix loop using the validate-flink-sql skill — not the Agno FlinkSqlDeployFixerAgent.

Requires FLINK_* credentials in repo .env and MCP enabled (Cursor) or flink-skill-mcp configured (Claude Code).

Cursor¶

Load validate-flink-sql. For failed statement {table}-dml, call MCP get_flink_statement_exceptions, fix DDL/DML using skill rules, re-validate with validate_flink_sql_offline, then redeploy source DDLs, target DDL, and target DML.

Claude Code¶

Load validate-flink-sql. Call MCP get_flink_statement_exceptions on the failed statement. Fix SQL using skill rules, run flink-skill-validate offline, then redeploy via MCP create_flink_statement and wait_flink_statement_phase.

Workflow:

1. get_flink_statement_exceptions on the failed statement name.
2. Apply validate-flink-sql rules; update ddl.{table}.sql, dml.{table}.sql, and tests/ddl.*.sql stubs if needed.
3. Re-validate (validate_flink_sql_offline or flink-skill-validate offline).
4. Redeploy in order: source stub DDLs → target DDL → target DML.
5. check_flink_statement_health on DML; repeat until success or you stop.

Do not set AGENT_FIXER_EXECUTION_ENABLED or invoke Agno harness CLIs for IDE fixes.

Scope 2: ksqlDB to Flink migration with CC validation¶

Use when converting ksqlDB CREATE STREAM / CREATE TABLE scripts to Flink DDL and DML.

Ksql routing examples from Confluent KSQL tutorial¶

The references/ksql/sources/routing includes the following ksql from the Confluent.io tutorial:

├── routing
│   ├── deduplicate.ksql
│   ├── filtering.ksql
│   ├── insert_acting_events.sql
│   ├── insert_clicks.sql
│   ├── insert_songs.sql
│   ├── merge.ksql
│   └── splitting.ksql

CLI translation only¶

cd ksql-to-flink-skill/harness && uv sync --extra dev

uv run ksql-flink-migrate \
  --table dim_all_songs \
  --file ../../references/ksql/sources/routing/merge.ksql \
  --out-dir ../../staging/ksql-lab-out \
  --skip-deploy

Validate generated output:

uv run --directory flink-skill-common/harness flink-skill-validate offline \
  --ddl ../../staging/ksql-lab-out/ddl.dim_all_songs.sql \
  --dml ../../staging/ksql-lab-out/dml.dim_all_songs.sql

Agno harness: translation and deployment with fixer agent (CI only)¶

Requires AGENT_FIXER_EXECUTION_ENABLED=1 in .env and a reachable LLM (SL_LLM_*).

AGENT_FIXER_EXECUTION_ENABLED=1 uv run ksql-flink-migrate \
  --table dim_all_songs \
  --file ../../references/ksql/sources/routing/merge.ksql \
  --out-dir ../../staging/ksql-lab-out

This invokes FlinkSqlDeployFixerAgent (Agno) on validation or deploy failure — not the Cursor/Claude IDE workflow.

Claude Code¶

From the references folder, enter a prompt like:

Load the ksql-to-flink skill. Migrate ksql/sources/routing/deduplicate.ksql for table detected_clicks. Write ddl.detected_clicks.sql and dml.detected_clicks.sql under staging/ksql-lab-out/. Then validate with flink-skill-validate offline.

Workflow:

1. Apply ksql-to-flink translation rules yourself (do not run ksql-flink-migrate).
2. Write ddl.{table}.sql and dml.{table}.sql under the output directory.
3. Run flink-skill-validate offline on the outputs (or MCP validate_flink_sql_offline in Cursor).
4. On errors, follow the validate-flink-sql fix loop (apply rules, re-validate).
5. Optional deploy: MCP tools from flink-skill-common (create_flink_statement, etc.) when Flink credentials are configured. On deploy failure, follow the same fix loop (Scope 1).

Cursor¶

Load the ksql-to-flink skill. Migrate ksql/sources/routing/deduplicate.ksql for table detected_clicks to staging/ksql-lab-out/. Validate with MCP validate_flink_sql_offline.

Same workflow as Claude Code, but use MCP tools for validation and deploy instead of shell CLIs.