A modular, batteries-included data engineering platform built for the complete data lifecycle.

Engine-agnostic by design. Every component has a default open-source implementation and a stable Python Protocol interface — swap any layer without touching pipeline code.

┌─────────────────────────────────────────────────────────────┐
│                    User Interface Layer                      │
│         CLI (odep)    │    Python SDK    │    Web UI         │
└──────────────────────────────┬──────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────┐
│              API Gateway (FastAPI)                           │
│     OIDC/OAuth2 Auth  │  Rate Limiting  │  Audit Logging    │
└──────────────────────────────┬──────────────────────────────┘
                               │
         ┌─────────────────────┼─────────────────────┐
         ▼                     ▼                     ▼
┌────────────────┐  ┌──────────────────┐  ┌──────────────────┐
│ Metadata Layer │  │Orchestration Layer│  │ Execution Layer  │
│  MetadataService│  │   Orchestrator   │  │ ExecutionEngine  │
│  Protocol      │  │   Protocol       │  │ Protocol         │
│                │  │                  │  │                  │
│  OpenMeta      │  │  Airflow ✓       │  │  DuckDB ✓        │
│  (DataHub +    │  │  Dagster         │  │  Spark           │
│   OpenLineage +│  │  Prefect         │  │  Flink           │
│   GE + OPA)    │  │  Temporal        │  │  dbt             │
└────────────────┘  └──────────────────┘  │  Trino           │
                                          │  ClickHouse      │
                                          │  MetaMind v2.0   │
                                          └──────────────────┘
                                                   │
                               ┌───────────────────▼──────────────────┐
                               │           Storage Layer               │
                               │  Lakehouse: Iceberg / Delta / Hudi    │
                               │  Formats:  Parquet / ORC / Avro / JSON│
                               │  Object:   MinIO / S3 / GCS / Azure   │
                               └──────────────────────────────────────┘

pip install odep
# or from source
git clone https://github.com/your-org/odep && cd odep
pip install -e ".[dev]"

odep init my-platform --engine=duckdb
cd my-platform

# Full stack: Airflow + DataHub + Marquez + Spark + MinIO (~8GB RAM)
odep local up

# Minimal stack: Marquez only (~1GB RAM)
odep local up --profile=minimal

odep deploy pipelines/user_events.yaml --env=local
odep run user_events_etl

# Deploy with a cron schedule
odep deploy pipelines/orders.yaml --env=prod --schedule="0 2 * * *"

# Backfill January 2024
odep run orders_etl --backfill --start=2024-01-01 --end=2024-02-01

# Stream last 200 log lines
odep logs run_abc123 --tail=200

# Run quality checks with LLM evaluation
odep test orders.csv --suite rules/orders.yaml --llm-evaluate

# Show upstream lineage
odep lineage "urn:li:dataset:(bigquery,prod.analytics.fact_orders,prod)"

# Switch execution engine
odep config set ODEP_EXECUTION__DEFAULT_ENGINE=metamind

from odep.adapters.duckdb.adapter import DuckDbAdapter
from odep.config import ExecutionConfig
from odep.models import JobConfig, EngineType

adapter = DuckDbAdapter(ExecutionConfig())
handle = adapter.submit(JobConfig(
    engine=EngineType.SQL,
    code="SELECT COUNT(*) FROM read_parquet('data/*.parquet')",
    dependencies=[], cluster_config={}, io_config={}
))
result = adapter.wait_for_completion(handle)
print(result.records_processed)

odep config set ODEP_EXECUTION__DEFAULT_ENGINE=spark
odep config set ODEP_EXECUTION__SPARK_MASTER=spark://localhost:7077

odep config set ODEP_EXECUTION__DEFAULT_ENGINE=flink
odep config set ODEP_EXECUTION__FLINK_JOBMANAGER_URL=http://localhost:8083

odep config set ODEP_EXECUTION__DEFAULT_ENGINE=dbt
odep config set ODEP_EXECUTION__DBT_PROJECT_DIR=./dbt_project

odep config set ODEP_EXECUTION__DEFAULT_ENGINE=trino
odep config set ODEP_EXECUTION__TRINO_HOST=localhost
odep config set ODEP_EXECUTION__TRINO_PORT=8082

odep config set ODEP_EXECUTION__DEFAULT_ENGINE=clickhouse
odep config set ODEP_EXECUTION__CLICKHOUSE_HOST=localhost

odep config set ODEP_ORCHESTRATION__ENGINE=airflow
odep config set ODEP_ORCHESTRATION__AIRFLOW_URL=http://localhost:8090

odep config set ODEP_ORCHESTRATION__ENGINE=dagster
odep config set ODEP_ORCHESTRATION__DAGSTER_URL=http://localhost:3000

odep config set ODEP_ORCHESTRATION__ENGINE=prefect
odep config set ODEP_ORCHESTRATION__PREFECT_URL=http://localhost:4200

odep config set ODEP_ORCHESTRATION__ENGINE=temporal
odep config set ODEP_ORCHESTRATION__TEMPORAL_HOST=localhost:7233

from odep.factory import get_metadata_adapter
from odep.config import OdepConfig
from odep.models import DatasetMetadata, LineageEdge

config = OdepConfig()
meta = get_metadata_adapter(config.metadata.engine, config.metadata)

# Register a dataset
urn = meta.register_dataset(DatasetMetadata(
    urn="urn:li:dataset:(bigquery,prod.analytics.fact_orders,prod)",
    name="fact_orders",
    platform="bigquery",
    env="prod",
    schema=[{"name": "order_id", "type": "STRING"}, {"name": "amount", "type": "FLOAT"}],
    owner="data-team",
))

# Record lineage
meta.create_lineage([LineageEdge(
    source_urn="urn:li:dataset:(bigquery,prod.raw.orders,prod)",
    target_urn=urn,
    transformation="SELECT order_id, SUM(amount) FROM raw.orders GROUP BY 1",
)])

# Traverse upstream
graph = meta.get_full_upstream(urn, max_depth=5)

# Governance
meta.apply_tag(urn, "PII")
meta.check_access("alice@company.com", urn, "read")  # → True/False

name: orders_quality
dataset_urn: "urn:li:dataset:(bigquery,prod.analytics.fact_orders,prod)"
rules:
  - name: order_id_not_null
    rule_type: not_null
    column: order_id
    severity: critical
    blocking: true

  - name: amount_positive
    rule_type: min
    column: amount
    params:
      min: 0.0
    severity: error
    blocking: true

  - name: status_valid
    rule_type: accepted_values
    column: status
    params:
      values: [pending, processing, shipped, delivered, cancelled]
    severity: warning
    blocking: false

  - name: data_freshness
    rule_type: freshness
    column: created_at
    params:
      max_age_hours: 25
    severity: error
    blocking: true

odep test orders.csv --suite rules/orders.yaml
# exits non-zero if any blocking rule fails

# Auto-generate rules from data
odep dq generate orders.csv -o rules/orders.yaml

from odep.dq.engine import NativeQualityEngine
from odep.dq.serializer import load_suite
import pandas as pd

engine = NativeQualityEngine()
suite = load_suite("rules/orders.yaml")
df = pd.read_csv("orders.csv")

result = engine.run_suite(suite, df)
print(f"Quality score: {result.quality_score:.1f}%")
print(f"Passed: {result.passed}/{result.total_rules}")

# Raise on blocking failure
engine.assert_suite(suite, df)

odep config set ODEP_EXECUTION__DEFAULT_ENGINE=metamind
odep config set ODEP_METAMIND__METAMIND_URL=http://localhost:8000
odep config set ODEP_METAMIND__TENANT_ID=my-tenant
odep config set ODEP_METAMIND__API_TOKEN=<your-token>

from odep.adapters.metamind.adapter import MetaMindAdapter
from odep.config import MetaMindConfig
from odep.models import JobConfig, EngineType

adapter = MetaMindAdapter(MetaMindConfig())
handle = adapter.submit(JobConfig(
    engine=EngineType.SQL,
    code="SELECT customer_id, SUM(amount) FROM orders GROUP BY 1",
    dependencies=[], cluster_config={}, io_config={}
))

metrics = adapter.get_metrics(handle)
# {
#   "optimization_tier": 1,          # Cascades depth
#   "cache_hit": True,               # Redis plan cache
#   "workload_type": "OLAP",         # Classified workload
#   "backend_used": "duckdb",        # Actual backend routed to
#   "optimization_ms": 12.4,         # Optimizer time
#   "plan_cost": 0.0034,             # Estimated cost
#   "flags_used": ["F09", "F24"]     # Active feature flags
# }

from odep.sdk import Pipeline, deploy_pipeline, execute_with_quality_gate
from odep.config import OdepConfig

config = OdepConfig()

# Load and validate a pipeline
pipeline = Pipeline.from_file("pipelines/orders.yaml")
pipeline.validate()  # local — no network calls

# Deploy
job_id = deploy_pipeline("pipelines/orders.yaml", env="prod", config=config)

# Execute with quality gate
result = execute_with_quality_gate(job_id, run_conf={}, config=config)
print(f"Processed {result.records_processed:,} records in {result.execution_time_ms}ms")

from odep.factory import get_metadata_adapter, get_orchestrator_adapter, get_execution_adapter
from odep.config import OdepConfig

config = OdepConfig()

meta = get_metadata_adapter("openmeta", config.metadata)
orch = get_orchestrator_adapter("airflow", config.orchestration)
exec = get_execution_adapter("duckdb", config.execution)
# or swap to MetaMind:
exec = get_execution_adapter("metamind", config.metamind)

# Start full stack
odep local up
# or directly:
docker compose --profile full up -d

# Start only Trino + Flink
docker compose --profile trino --profile flink up -d

# Stop everything and remove volumes
odep local down --volumes

# .odep.env
ODEP_METADATA__ENGINE=openmeta
ODEP_METADATA__DATAHUB_URL=http://localhost:8080
ODEP_METADATA__MARQUEZ_URL=http://localhost:5000

ODEP_ORCHESTRATION__ENGINE=airflow
ODEP_ORCHESTRATION__AIRFLOW_URL=http://localhost:8090
ODEP_ORCHESTRATION__AIRFLOW_USERNAME=admin
ODEP_ORCHESTRATION__AIRFLOW_PASSWORD=admin

ODEP_EXECUTION__DEFAULT_ENGINE=duckdb
ODEP_EXECUTION__SPARK_MASTER=local[*]
ODEP_EXECUTION__TRINO_HOST=localhost
ODEP_EXECUTION__TRINO_PORT=8082

ODEP_METAMIND__METAMIND_URL=http://localhost:8000
ODEP_METAMIND__TENANT_ID=default
ODEP_METAMIND__API_TOKEN=

cd infra/terraform/aws
terraform init
terraform apply \
  -var="metadata_engine=openmeta" \
  -var="orchestration_engine=airflow" \
  -var="environment=prod"

terraform output metadata_endpoint
# → https://datahub.internal.company.com

odep template list
# batch-pipeline        Standard batch ETL with DuckDB/Spark
# streaming-pipeline    Kafka + Flink streaming
# ml-feature-pipeline   Feature engineering with quality gates
# dbt-project           dbt project with ODEP wrapper

odep template use batch-pipeline --name=user_events_etl

# Prometheus metrics
curl http://localhost:8000/metrics

# Grafana dashboards are included in infra/monitoring/grafana/

# Unit tests
pytest tests/unit/ -v

# Property-based tests (hypothesis)
pytest tests/property/ -v

# Integration tests (requires Docker)
pytest tests/integration/ -v

# Full suite with coverage
pytest --cov=odep --cov-report=term-missing

odep/
├── interfaces/          # Python Protocol definitions
│   └── __init__.py      # MetadataService, Orchestrator, ExecutionEngine
├── adapters/
│   ├── openmeta/        # MetadataService — DataHub + OpenLineage + GE + OPA
│   ├── airflow/         # Orchestrator — Apache Airflow REST API
│   ├── dagster/         # Orchestrator — Dagster GraphQL API
│   ├── prefect/         # Orchestrator — Prefect REST API
│   ├── temporal/        # Orchestrator — Temporal SDK
│   ├── duckdb/          # ExecutionEngine — local DuckDB
│   ├── spark/           # ExecutionEngine — Apache Spark
│   ├── flink/           # ExecutionEngine — Apache Flink SQL Gateway
│   ├── dbt/             # ExecutionEngine — dbt Core
│   ├── trino/           # ExecutionEngine — Trino REST API
│   ├── clickhouse/      # ExecutionEngine — ClickHouse HTTP API
│   └── metamind/        # ExecutionEngine — MetaMind v2.0 REST API
├── dq/                  # Native data quality engine
│   ├── engine.py        # NativeQualityEngine (13 rule types)
│   ├── models.py        # QualityRule, QualitySuite, SuiteResult
│   ├── runner.py        # Suite runner with quality gate
│   └── serializer.py    # YAML suite loader/writer
├── sdk/
│   ├── pipeline.py      # Pipeline.from_file(), validate(), extract_lineage_edges()
│   ├── deploy.py        # deploy_pipeline()
│   └── execute.py       # execute_with_quality_gate()
├── cli/
│   ├── main.py          # odep CLI entry point
│   └── commands/        # init, local, template, config, dq
├── api/
│   ├── auth.py          # OIDC/OAuth2 JWT middleware
│   ├── middleware.py     # Rate limiting
│   ├── audit.py         # OpenTelemetry audit logging
│   └── routes/          # FastAPI routers
├── ui/                  # React web console
│   └── src/pages/       # Catalog, Lineage, Quality, Orchestration, Execution
├── config.py            # OdepConfig (Pydantic settings)
├── factory.py           # AdapterFactory
├── models.py            # DatasetMetadata, LineageEdge, JobDefinition, JobResult
└── exceptions.py        # Typed exceptions

infra/
├── terraform/
│   ├── aws/             # AWS deployment module
│   ├── gcp/             # GCP deployment module
│   └── azure/           # Azure deployment module
└── monitoring/
    └── grafana/         # Grafana dashboards

tests/
├── unit/                # Unit tests per adapter
├── property/            # hypothesis property-based tests (P1–P16)
└── integration/         # TestContainers integration tests

docker-compose.yml       # Local dev stack (profiles: full, minimal, spark, trino, flink, ...)
pyproject.toml           # Package metadata and dependencies