dagster-floe 0.2.5

Dagster connector for Floe CLI (config-driven ingestion)

Floe + Dagster

This folder contains the Dagster connector for Floe.

For local setup of both Dagster and Airflow with isolated virtual environments, see:

• orchestrators/LOCAL_DEV.md

Current model:

• Parse-time reads Floe manifests (floe.manifest.v1), not Floe YAML.
• Supports single manifest and multi-manifest directory loading.
• Builds one asset per entity and one job per manifest.
• Runs Floe with manifest execution contract and JSON logs.
• Publishes run/entity metadata from NDJSON + summary.

What Is Implemented

• Manifest-first orchestration (floe.manifest.v1).
• Strict schema validation at manifest load time.
• Asset generation from entities[] (asset_key, group_name respected).
• One Dagster job per manifest (stable job names).
• Multi-manifest loading (*.manifest.json) with collision checks.
• Local runner (local_process) support.
• execution.defaults.env and execution.defaults.workdir support.
• Floe quality outcomes exposed as Dagster native Asset Checks (cast_error, not_null, unique, schema_mismatch, file_status) from Floe reports.

Install

Prereqs:

• Floe installed (either the floe CLI binary or Docker with a Floe image).
• Python 3.10+.

pip install dagster-floe

Development install (from this repo)

python3 -m venv orchestrators/dagster-floe/.venv
source orchestrators/dagster-floe/.venv/bin/activate
pip install -e orchestrators/dagster-floe[dev]

Generate a manifest

floe manifest generate \
  -c orchestrators/dagster-floe/example/config.yml \
  --output orchestrators/dagster-floe/example/manifest.dagster.json

Run the example (repo-only)

cd orchestrators/dagster-floe
FLOE_MANIFEST_DIR=./example/manifests dagster dev

The example workspace loads example/definitions.py, which wires local example files/manifest to the reusable connector APIs. The repository example includes two manifests by domain:

• example/manifests/hr.manifest.json
• example/manifests/sales.manifest.json

Asset checks

Every accepted-output asset automatically gets quality checks registered in Dagster — no extra config needed. After each run, the connector reads the Floe run report and publishes pass/fail results for each entity.

Check	Fails when
file_status	One or more input files had a failed or error-level processing status (e.g., unreadable file, parse failure)
cast_error	Type-casting failures exceeded the entity's policy threshold
not_null	Null values found in non-nullable columns
unique	Duplicate values found in uniqueness-constrained columns
schema_mismatch	Incoming file schema is incompatible with the declared schema

Checks appear in the Dagster UI under each asset, linked to the run that produced them. A WARN-severity violation marks the check as a warning; REJECT or ABORT marks it as a failure.

Lineage integration

The connector automatically injects two environment variables into every floe subprocess:

• DAGSTER_RUN_ID — the current Dagster run ID
• DAGSTER_JOB_NAME — the Dagster job name

When lineage is enabled and both variables are present, floe attaches a parent facet to its OpenLineage run event. This facet declares that the Floe run is a child of the Dagster run, making the lineage graph show the full Dagster → Floe pipeline hierarchy.

The Floe run event's own job.name is derived from lineage.job_name in the embedded manifest lineage config, or from the config file stem. To control it explicitly, add job_name to your lineage block when generating the manifest:

# prod.yml (profile)
lineage:
  url: "http://marquez:5000"
  namespace: "my-data-platform"
  job_name: "orders-pipeline"   # sets job.name on the Floe OpenLineage run event

floe manifest generate -c orders.yml -p prod.yml --output manifests/orders.json

See docs/lineage.md for the full lineage config reference.

Notes

• This connector does not parse YAML directly; it consumes floe.manifest.v1.
• Connector logic lives under src/floe_dagster/; local wiring for demo lives in example/definitions.py.
• For local development without an installed floe binary, you can point LocalRunner to a custom command, e.g.:
  • LocalRunner(\"cargo run -p floe-cli --\")
• Manifest runner support in connector is currently local_process only.
• For local setup commands, use orchestrators/LOCAL_DEV.md.
• Design notes and future work: orchestrators/dagster-floe/INTEGRATION_SPEC.md

What Is Not Implemented Yet

• Kubernetes/ECS runner adapters.
• Cloud summary loading (s3://, gs://, abfs://).
• Single-process multi-entity fan-out execution mode.

Releasing

This repo is a monorepo. Floe and this connector are versioned and tagged independently:

• Floe CLI release tags: vX.Y.Z
• Dagster connector release tags: dagster-floe-vX.Y.Z (triggers the PyPI publish workflow)

Example:

git checkout main
git pull
git tag dagster-floe-v0.1.0
git push origin dagster-floe-v0.1.0