ATC/atc-agents
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(dataflow): clean left-to-right stage layout for the topology

Browse Source 
Align nodes into pipeline columns (producers -> sources -> CDC -> lakehouse ->
Trino) with governance centred at the bottom, so lineage reads in order instead
of Iceberg/Trino floating mid-canvas.
This commit is contained in:
mo
2026-06-27 12:07:25 +02:00
parent 05794906a7
commit ca05e04f8c
1 changed files with 20 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/dataflow.py
+20 -11
View File
@@ -26,19 +26,28 @@ TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/")
TRINO_USER = os.getenv("TRINO_USER", "mo")
# Static landscape (x,y in 0..100). kind drives UI styling.
# Laid out as clean left→right pipeline stages so lineage reads in order:
# producers (col 0) → sources (col 1) → CDC (col 2) → storage/lakehouse (col 3)
# → query engine (col 4); governance (OpenMetadata) sits centred at the bottom.
NODES: list[dict[str, Any]] = [
{"id": "generator", "label": "Data Generator", "sub": "Airflow DAGs", "kind": "generator", "x": 10, "y": 50},
{"id": "hdfs", "label": "Hadoop HDFS", "sub": "historical_sales", "kind": "hadoop", "x": 10, "y": 88},
{"id": "postgres", "label": "PostgreSQL", "sub": "sales_orders", "kind": "source", "x": 30, "y": 18},
{"id": "mysql", "label": "MySQL", "sub": "employee_events", "kind": "source", "x": 30, "y": 44},
{"id": "mongodb", "label": "MongoDB", "sub": "events", "kind": "source", "x": 30, "y": 70},
{"id": "kafka", "label": "Kafka · Debezium", "sub": "CDC topics", "kind": "stream", "x": 52, "y": 40},
{"id": "s3_cdc", "label": "S3 CDC Archive", "sub": "object store", "kind": "sink", "x": 72, "y": 20},
{"id": "iceberg_hadoop", "label": "Iceberg · hadoop", "sub": "historical_sales_hdfs", "kind": "lakehouse", "x": 72, "y": 60},
{"id": "iceberg_curated", "label": "Iceberg · curated_masked", "sub": "masked PII", "kind": "lakehouse", "x": 72, "y": 86},
{"id": "trino", "label": "Trino", "sub": "query engine", "kind": "engine", "x": 90, "y": 52},
# col 0 — producers / origin
{"id": "generator", "label": "Data Generator", "sub": "Airflow DAGs", "kind": "generator", "x": 9, "y": 34},
{"id": "hdfs", "label": "Hadoop HDFS", "sub": "historical_sales", "kind": "hadoop", "x": 9, "y": 80},
# col 1 — source databases
{"id": "postgres", "label": "PostgreSQL", "sub": "sales_orders", "kind": "source", "x": 28, "y": 16},
{"id": "mysql", "label": "MySQL", "sub": "employee_events", "kind": "source", "x": 28, "y": 38},
{"id": "mongodb", "label": "MongoDB", "sub": "events", "kind": "source", "x": 28, "y": 60},
# col 2 — change data capture
{"id": "kafka", "label": "Kafka · Debezium", "sub": "CDC topics", "kind": "stream", "x": 47, "y": 34},
# col 3 — storage / lakehouse
{"id": "s3_cdc", "label": "S3 CDC Archive", "sub": "object store", "kind": "sink", "x": 67, "y": 13},
{"id": "iceberg_curated", "label": "Iceberg · curated_masked", "sub": "masked PII", "kind": "lakehouse", "x": 67, "y": 45},
{"id": "iceberg_hadoop", "label": "Iceberg · hadoop", "sub": "historical_sales_hdfs", "kind": "lakehouse", "x": 67, "y": 80},
# col 4 — query engine
{"id": "trino", "label": "Trino", "sub": "query engine", "kind": "engine", "x": 88, "y": 45},
# governance — bottom centre
{"id": "openmetadata", "label": "OpenMetadata", "sub": "catalog · lineage · PII", "kind": "governance",
"x": 50, "y": 92, "url": "http://10.0.21.47:8585"},
"x": 47, "y": 93, "url": "http://10.0.21.47:8585"},
]
# Edges. movement_id (optional) links to movements.py so the edge is triggerable.