{"id":"31d92dd4-48b6-4a46-b6b3-941d07bd18bc","slug":"databricks-solutions-flowx","kind":"skill","name":"flowx","summary":"Automate migrations to Databricks Lakeflow Jobs","owner":"databricks-solutions","repoUrl":"https://github.com/databricks-solutions/flowx","license":"Apache-2.0","licenseTier":"permissive","trust":"official","stars":6,"match":null,"fewStars":false,"format":"skill","createdAt":"2026-06-15T19:31:43.000Z","addedAt":"2026-10-21T17:33:01.404Z","logoUrl":"https://avatars.githubusercontent.com/u/168765251?v=4","linkOnly":false,"collection":null,"skillPath":null,"categories":["writing"],"tags":[],"runtime":"binary","local":true,"cloudCapable":false,"cloudCapability":"unknown","cloudReadyOnPlatform":false,"needsApiKey":false,"reviewed":false,"syncedAt":"2026-10-25T11:56:16.079Z","sample":false,"url":"https://conduits.dev/tools/databricks-solutions-flowx","language":"Python","transports":["stdio"],"install":{},"requirements":{"env":[],"upstreamMayCharge":false},"tools":null,"readme":"# flowx\n\nOrchestrator-to-Databricks Lakeflow Jobs translator, delivered as agent skills.\n\nflowx converts a source orchestrator's pipelines — **Azure Data Factory (ADF)** or **Apache\nAirflow** — into Databricks Lakeflow Jobs packaged as Declarative Automation Bundles (DABs). It\ndeterministically translates known activity/operator types and falls back to agentic (LLM-assisted)\ntranslation for complex or rare types. flowx runs as a set of [agent skills](skills/) usable from\nDatabricks Genie Code, Claude Code, or any tool that supports the Agent Skills standard.\n\nBoth sources emit the same source-neutral Pipeline IR, so the convert-configuration and package\nphases are shared; only discovery and translation are source-specific. Pick the source with\n`--source {adf,airflow}` (required for discover/convert; package is source-independent).\n\n## Architecture\n\n```\n                         flowx Pipeline\n                         ==================\n\n  ADF ARM/JSON (UC Volumes / Workspace)   |   Airflow DAG .py files\n                          \\               |              /\n                           v              v             v\n  +---------------------------------------------------------------+\n  |  1. DISCOVER   sources/<adf|airflow>/  -> metadata/inventory.json\n  |                (ADF: ARM/JSON parse; Airflow: static ast parse)\n  +---------------------------------------------------------------+\n        |\n        v\n  +---------------------------------------------------------------+\n  |  2. CONVERT    sources/<adf|airflow>/  -> shared Pipeline IR\n  |                (deterministic mappings + agentic gaps)\n  +---------------------------------------------------------------+\n        |\n        v\n  +---------------------------------------------------------------+\n  |  3. PACKAGE    bundler/dab_writer.py (source-independent)\n  |                IR -> DAB YAML + notebooks + setup scripts\n  +---------------------------------------------------------------+\n        |\n        v\n  databricks bundle validate / deploy\n```\n\nThe phases are exposed two ways: as **skills** the agent runs directly (via a Python virtual\nenvironment locally), and as a single **MCP tool** hosted on a Databricks App (for Genie Code). See\n[Running flowx as an MCP server](#running-flowx-as-an-mcp-server).\n\n## Installation\n\nflowx installs in one of two shapes depending on where your agent runs. Full, step-by-step\ninstructions for both are in the [installation docs](docs/content/docs/installation.mdx); the\nsummary:\n\n### Databricks Genie Code\n\nThe phases run as an MCP server (a Databricks App). Clone the repo into **`/Workspace/Shared`**\n(so the app's service principal can read the source), copy `skills/` into your skills folder, then\nrun the setup skill:\n\n```\n@flowx-setup\n```\n\nOn Databricks, `flowx-setup` deploys the `mcp-flowx` app for you. You can also deploy it directly by\nrunning the **`app/deploy_app.py`** notebook (SDK-based, works on serverless), or `app/deploy.sh`\nfrom a workspace web terminal. Then add the app under Genie Code **Settings → MCP Servers → Add\nServer → Custom MCP server**.\n\n### Claude Code (and other local agent harnesses)\n\nflowx is a Claude Code plugin distributed through its marketplace:\n\n```\n/plugin marketplace add databricks-solutions/flowx\n/plugin install flowx@flowx\n```\n\nRun `/reload-plugins`, then set up the local runtime once:\n\n```\n/flowx:flowx-setup\n```\n\nThis provisions a Python virtual environment (via `scripts/bootstrap.sh`) and writes a\n`.migration-venv` marker the phase skills read. No `uv` is required for plugin users.\n\n## Usage\n\nRun the end-to-end migration:\n\n```\n/flowx:flowx-migrate\n```\n\nOr run individual phases:\n\n```\n/flowx:flowx-extract-metadata  # (ADF) Get the factory's definitions: ADF Studio export, Git, or live Azure API\n/flowx:flowx-discover    # Parse the source (ADF JSON / Airflow DAGs), produce inventory + complexity report\n/flowx:flowx-convert     # Deterministic + agentic translation\n/flowx:flowx-package     # Generate DABs project\n/flowx:flowx-profile     # (ADF, optional) Estimate current ADF spend from a live Azure scan\n```\n\n(In Genie Code, invoke the same skills with the `@` prefix, e.g. `@flowx-migrate`.)\n\n## Setup\n\n`/flowx:flowx-setup` keys off the `DATABRICKS_RUNTIME_VERSION` environment variable\n(the same signal the rest of the plugin uses to detect Databricks) and prepares one\nof two execution paths:\n\n- **Local / Claude Code (virtual environment).** The phases run from the plugin's\n  CLI. Setup runs `scripts/bootstrap.sh`, which creates a `.venv`, installs\n  `requirements.txt`, and writes the resolved interpreter path to a\n  `.migration-venv` marker file that the phase skills read. Optionally, a local\n  (stdio) MCP server can be registered to drive the phases through MCP tools\n  instead of the CLI.\n\n- **Databricks Genie Code (MCP server, no virtual environment).** The phases run as\n  a single `flowx` MCP tool hosted on a Databricks App. Setup runs `app/deploy.sh`,\n  which stages a self-contained bundle, syncs it to `/Workspace/Shared/mcp-flowx`,\n  and deploys the `mcp-flowx` app. You then grant app/data access and register the\n  app under Genie Code **Settings → MCP Servers**. No venv is created on this path.\n\nRun setup once before any other flowx skill, or again whenever the environment is\nmissing.\n\n## Supported ADF Activity Types\n\n### Deterministic (16 types)\n\n| ADF Activity | Databricks Task | Category |\n|---|---|---|\n| Copy | Notebook task | Data movement |\n| DatabricksNotebook | Notebook task | Compute |\n| DatabricksSparkJar | Spark JAR task | Compute |\n| DatabricksSparkPython | Spark Python task | Compute |\n| ForEach | for_each_task | Control flow |\n| IfCondition | if_else_task | Control flow |\n| Switch | if_else_task chain | Control flow |\n| SetVariable | run_job_task | Control flow |\n| AppendVariable | run_job_task | Control flow |\n| Filter | Notebook task | Control flow |\n| Wait | Notebook task (sleep) | Control flow |\n| Lookup | Notebook task | Data access |\n| WebActivity | Notebook task | External |\n| Delete | Notebook task | Data management |\n| ExecutePipeline | run_job_task | Orchestration |\n| DatabricksJob | run_job_task | Compute |\n\n### Agentic Fallback (12 types)\n\nActivities with complex semantics, or without a direct Databricks equivalent, are translated by the\nagent using LLM-assisted reasoning from the activity's ARM JSON.\n\n| ADF Activity | Strategy |\n|---|---|\n| ExecuteDataFlow | LLM-assisted (agentic) |\n| SqlServerStoredProcedure | LLM-assisted (agentic) |\n| AzureFunction | LLM-assisted (agentic) |\n| WebHook | LLM-assisted (agentic) |\n| Custom | LLM-assisted (agentic) |\n| ExecuteSSISPackage | LLM-assisted (agentic) |\n| AzureMLExecutePipeline | LLM-assisted (agentic) |\n| GetMetadata | LLM-assisted (agentic) |\n| Validation | LLM-assisted (agentic) |\n| Fail | LLM-assisted (agentic) |\n| Script | LLM-assisted (agentic) |\n| Until | LLM-assisted (agentic) |\n\n## Supported Airflow Operators\n\nThe Airflow source parses DAG `.py` modules **statically** (via `ast`, no Airflow install or DAG execution) and maps the constructs below to the shared IR — 74 operator/sensor types in the parser registry, plus the dbt CLI, Cosmos, and TaskFlow constructs.\n\nUnlike the ADF path, an Airflow operator's path is decided **per instance** by whether its inputs are statically resolvable. A recognized operator lowers to a real Databricks task (**deterministic**), or — when a value can't be resolved from source alone — to a failing placeholder + `gaps.json` entry (**agentic**). A few constructs are always agentic because their behavior is inherently runtime.\n\n| Category | Operator(s) | Databricks target | Path |\n|---|---|---|---|\n| Compute / scripts | `PythonOperator` | Notebook task (callable + transitive deps) | Deterministic † |\n| Compute / scripts | `PythonVirtualenvOperator`, `ExternalPythonOperator` | Notebook task + `%pip install` | Deterministic |\n| Compute / scripts | `BashOperator`, `SSHOperator` | `%sh` notebook (`spark-submit` lifted when option arity is known) | Deterministic † |\n| Compute / scripts | `SparkSubmitOperator` | Spark JAR / Python task | Deterministic |\n| Teradata | `BteqOperator`, `TeradataOperator`, and Bash/SSH commands that run `bteq`, TPT (`tbuild`, `tdload`), or the load/export utilities | Placeholder + gap carrying the script | Agentic |\n| Databricks provider | `DatabricksSubmitRunOperator` (+ `…DeferrableOperator`) | Notebook / run task | Deterministic |\n| Databricks provider | `DatabricksRunNowOperator` (+ `…DeferrableOperator`) | `run_job_task` | Deterministic |\n| Databricks provider | `DatabricksNotebookOperator` | Notebook task | Deterministic |\n| SQL | `DatabricksSqlOperator`, `DatabricksSQLStatementsOperator`, `SQLExecuteQueryOperator`, `PostgresOperator`, `MySqlOperator`, `HiveOperator` | `sql_task` | Deterministic † |\n| SQL | `DatabricksCopyIntoOperator` | `sql_task` (COPY INTO) | Deterministic |\n| Orchestration | `TriggerDagRunOperator` | `run_job_task` (target DAG by sanitized job name) | Deterministic |\n| Control flow | `BranchPythonOperator`, `ShortCircuitOperator` | Placeholder + gap | Agentic |\n| Control flow | `DummyOperator`, `EmptyOperator` | — (dropped, dependencies rewired) | Dropped |\n| Notification | `EmailOperator` | Placeholder + gap | Agentic |\n| Sensors — file | `S3KeySensor`, `GCSObjectExistenceSensor`, `FileSensor`, `HdfsSensor`, `WebHdfsSensor` | `file_arrival` trigger (root, no schedule) or `dbutils.fs` polling task | Deterministic † |\n| Sensors — table/SQL | `DatabricksPartitionSensor`, `DatabricksSqlSensor`, `DatabricksSQLStatementsSensor`, `SqlSensor` | `table_update` trigger (root literal table, no schedule) or `spark.sql` polling task | Deterministic † |\n| Sensors — poll | `HttpSensor`, `PythonSensor`, `DateTimeSensor` (+ `…Async`) | Polling notebook task | Deterministic † |\n| Sensors — cross-DAG | `ExternalTaskSensor` (+ `…Async`) | Placeholder + gap | Agentic |\n| Sensors — time | `TimeSensor`, `TimeDeltaSensor` | Placeholder + gap | Agentic |\n| dbt | `DbtRun/Test/Seed/Snapshot/Build/DepsOperator`, Cosmos `DbtDag` / `DbtTaskGroup` | single `DbtFactoryActivity` (static explosion, or PyDABs via `--dbt-mode pydabs`) | Deterministic |\n| Dataproc / Managed Spark | `DataprocSubmitJobOperator`, `DataprocCreateBatchOperator` (+ `ManagedSpark…` aliases) with `pyspark_job` / `pyspark_batch` | `spark_python_task` | Deterministic † |\n| Dataproc / Managed Spark | same, with `spark_job` / `spark_batch` | `spark_jar_task` (JARs as task libraries) | Deterministic † |\n| Dataproc / Managed Spark | same, with inline `spark_sql_job` | `sql_task` | Deterministic † |\n| Dataproc / Managed Spark | same, with `spark_r_*`, `pyspark_notebook_batch`, `hive_job`, `hadoop_job`, `pig_job`, `flink_job`, `presto_job`, `trino_job` | Placeholder + gap | Agentic |\n| Dataproc / Managed Spark | cluster create / delete / start / stop, `DataprocJobSensor`, `DataprocBatchSensor` | — (absorbed into Jobs compute / native task completion, dependencies rewired) | Dropped ‡ |\n| Dataproc / Managed Spark | cluster update / scale / diagnose, workflow templates, batch control, cancel operation | Placeholder + gap | Agentic |\n| TaskFlow API | `@dag`, `@task`, `@task.virtualenv` | task; implicit XCom → `dbutils.jobs.taskValues` | Deterministic † |\n| TaskFlow API | `@task.expand([literal])` | `for_each_task` | Deterministic † |\n| TaskFlow API | `@task.branch`, `@task.short_circuit`, `@task_group` (decorator form) | Placeholder + gap | Agentic |\n\n**†** Deterministic when the relevant values are statically resolvable; otherwise the instance routes\nto an agentic gap — e.g. a non-literal command / SQL / endpoint / path, a SQL template file (`.sql` / `.hql`) that is not next to the DAG or in a static `template_searchpath`, a callable that reads Airflow\ntask context or XCom, an unsafe inline template context, or a non-literal `.expand()` /\n`.partial().expand()`.\n**Agentic** emits a failing placeholder + `gaps.json` entry (eligible for the leaf-gap\nresolver) because the construct's behavior is inherently runtime.\n**‡** Dropped only when provably safe: every job on the cluster migrates, no other task reads the removed\ntask's output, the sensor pairs with an upstream asynchronous submission (or matching `batch_id`) (its timing is reported, not turned into a task timeout), and rewiring does\nnot change trigger-rule behavior. Otherwise the task stays as a placeholder with the reason attached.\n\n**Scheduling & semantics** (source-level, applied alongside the operators above): cron → Quartz,\n`timedelta` → a `start_date`-anchored Quartz cron when its length divides an hour or a day (else periodic), `@continuous` → continuous mode, Airflow-3 Asset/Dataset lists →\n`ALL_UPDATED` / `ANY_UPDATED` table triggers, `trigger_rule` → `run_if`, `params={...}` / `Param` →\njob parameters, and `>>` / `<<` / `set_upstream` / `set_downstream` / TaskGroup edges → task\ndependencies. Interval macros (`{{ ds }}`, `{{ ts }}`, `{{ data_interval_start }}`, ...) are computed per run\nby a generated first task that reproduces Airflow's rendering: the previous schedule tick for an Airflow 2\ndata-interval timetable, the fire time for an Airflow 3 raw cron, and the trigger time for a manual run.\n\nOperators without a deterministic mapping become a failing placeholder and are recorded in\n`gaps.json` for review. Eligible leaf gaps can use the fingerprint-bound resolver backed by the pinned [`airflow-to-dabs`](https://github.com/park-peter/airflow-to-dabs/tree/main/providers/flowx-gap-resolver) provider profile; flowx retains ownership of parsing, graph identity, policy, IR, and packaging. Full matrix:\n[`skills/flowx-convert/sources/airflow-coverage.md`](skills/flowx-convert/sources/airflow-coverage.md).\n\nAirflow discovery independently audits DAG declarations, task candidates, dependency declarations,\nDAG settings, mapped calls, and operator arguments before comparing them with captured IR. An\nincluded DAG is `verified` when every audited construct has a proven translation,\n`verified_with_gaps` when every unsupported construct is linked to a runnable-failure placeholder,\nor `failed` when reconciliation finds unexplained loss. Failed reconciliation exits nonzero and\nblocks package writes. `--exclude-dag <dag_id>` is repeatable; excluded DAGs emit no Job but remain\nvisible with zero translated activities in inventory and coverage reporting. This guarantee applies\nto the supported static subset; flowx never imports or executes DAG modules.\n\n## How It Works\n\n### Phase 1: Discover\nParses the source into typed nodes and classifies each activity/operator as deterministic, agentic, or unsupported — ADF JSON from Unity Catalog volumes (or a `/Workspace` Git folder, normalizing ARM template format), or Airflow DAG `.py` modules read statically with `ast`. Airflow inventory includes audited/deterministic/agentic/failed/excluded counts, reconciliation status, stable finding fingerprints, translation-path coverage, and deterministic coverage. Produces `metadata/inventory.json` and a per-pipeline complexity report at `metadata/profile_report.csv`.\n\n### Phase 2: Convert\nApplies deterministic translators (ADF activity registry / Airflow operator mapping), resolves dependencies, and records unresolved gaps. ADF supports its guided agentic translation workflow. Airflow supports a fingerprint-bound, explicitly reviewed leaf-gap workflow whose constrained provider output is replayed against an immutable deterministic baseline before packaging. Produces the shared Pipeline IR consumed unchanged by the package phase.\n\n### Phase 3: Package\nConverts Pipeline IR into a deployable DABs project: `databricks.yml`, per-job YAML resource files, generated Python notebooks, and setup scripts for UC volumes, secrets, and connections. A multi-pipeline factory can be laid out into bundles with `--packaging-mode` (`per-pipeline` default / `single` / `per-group`); a top-level `DEPLOY.md` records the suggested callees-first deploy order, and `python -m flowx.adapter deploy` deploys the bundles in that order, wiring cross-bundle job ids automatically.\n\n## Output Format\n\nAll three phases write into one shared output directory (default `./flowx_output`):\n\n```\nflowx_output/\n  databricks.yml              # Bundle configuration (package)\n  resources/\n    jobs/\n      <pipeline_name>.yml     # One Job per included ADF pipeline or Airflow DAG\n  src/\n    notebooks/\n      <pipeline_name>/\n        <activity_name>.py    # Generated notebooks per activity\n    setup/\n      create_volumes.py       # UC volume setup\n      create_secrets.py       # Secret scope setup\n      create_connections.py   # Connection setup\n  SETUP.md                    # Setup instructions (package)\n  metadata/\n    inventory.json            # discover: activity inventory\n    profile_report.csv        # discover: per-pipeline complexity report\n    <pipeline>.arm.json       # discover: verbatim original ADF/ARM source\n    configuration.json        # modify: collected configuration answers\n  .work/                      # transient intermediates (translation report, IR, gaps.json); pruned by package\n```\n\n## Running flowx as an MCP server\n\nThe phases are also packaged as [Model Context Protocol](https://modelcontextprotocol.io) tools (in\n[`src/flowx/mcp/`](src/flowx/mcp)) so an agent can invoke them directly instead of shelling out to\nthe CLI. The server exposes a single `flowx(command, parameters)` tool to stay under host tool\nlimits. For Databricks Genie Code it runs as a Databricks App; see the [app README](app/README.md)\nfor deployment (SDK notebook or CLI script) and Genie Code registration.\n\n## Development\n\n```bash\nmake dev               # Install dependencies (uses uv)\nmake test              # Run unit tests\nmake integration       # Run integration tests (excludes the live-Azure suite; gates CI)\nmake integration-live  # Also run tests marked integration (live Azure; none today)\nmake fmt               # Format + lint (ruff + mypy)\nmake clean             # Remove build artifacts\n```\n\n### Prerequisites\n- Python 3.12+\n- [uv](https://docs.astral.sh/uv/) package manager\n\nThese prerequisites are for contributing to the flowx project. Plugin *users* do not need `uv` —\n`flowx-setup` provisions the runtime (a pip-based `.venv` locally, or the MCP server on Databricks).\n\n## Contributing\n\n1. Fork the repository\n2. Create a feature branch (`git checkout -b feature/my-feature`)\n3. Follow the [adding a new translator](CLAUDE.md#adding-a-new-deterministic-translator) guide\n4. Run `make fmt && make test` before committing\n5. Open a pull request\n"}