Architecture Decision Records (12)
Title
Centralized YAML Configuration with Typed Python Loader and Justfile Bridge
Description
Establish a centralized configuration architecture using structured YAML files under config/, parsed exclusively by a single type-safe Python loader (config_loader.py), with a lightweight justfile bridge to dynamically inject ports into Docker Compose at runtime.
Context
The Problem: Monolithic .env Wall of Text
A common pattern in data pipelines is to dump every single parameter into a monolithic, flat .env file at the project root. In real-world data platforms, this leads to massive configuration bloat that is extremely difficult to read, maintain, or secure:
# Example of monolithic configuration anti-pattern in large data projects
PROJECT_NAME=omnichannel-commerce-data-platform
ENVIRONMENT=dev
LOG_LEVEL=INFO
# Local PostgreSQL warehouse / raw landing
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=commerce_platform
POSTGRES_USER=commerce
POSTGRES_PASSWORD=commerce
POSTGRES_RAW_SCHEMA=raw
POSTGRES_STAGING_SCHEMA=staging
POSTGRES_MARTS_SCHEMA=marts
# Olist batch source
OLIST_DATASET_ID=olistbr/brazilian-ecommerce
OLIST_LANDING_PATH=storage/bronze/olist
KAGGLE_USERNAME=TODO_KAGGLE_USERNAME
KAGGLE_KEY=TODO_KAGGLE_KEY
# Retailrocket replay source
RETAILROCKET_DATASET_ID=retailrocket/ecommerce-dataset
RETAILROCKET_REPLAY_PATH=data/sample/streaming/retailrocket_events.jsonl
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_TOPIC_RETAILROCKET_RAW=retailrocket.events.raw
KAFKA_TOPIC_RETAILROCKET_VIEW=retailrocket.events.view
KAFKA_TOPIC_RETAILROCKET_ADDTOCART=retailrocket.events.addtocart
KAFKA_TOPIC_RETAILROCKET_TRANSACTION=retailrocket.events.transaction
KAFKA_TOPIC_RETAILROCKET_DLQ=retailrocket.events.dlq
KAFKA_CONSUMER_GROUP=retailrocket-replay-local
KAFKA_TOPIC_PARTITIONS=6
# MongoDB raw JSON / event document store
MONGO_URI=mongodb://localhost:27017
MONGO_DATABASE=commerce_raw
MONGO_COLLECTION_RETAILROCKET=retailrocket_events_raw
MONGO_COLLECTION_OPEN_FOOD_FACTS_PRODUCTS=open_food_facts_products_raw
MONGO_COLLECTION_WEATHER=open_meteo_weather_raw
MONGO_COLLECTION_FX=frankfurter_fx_raw
MONGO_COLLECTION_BATCH_RUNS=batch_api_runs_raw
# MinIO object storage
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=minio
MINIO_SECRET_KEY=minio123
MINIO_BUCKET_RAW=raw
MINIO_BUCKET_PROCESSED=processed
# Warehouse
WAREHOUSE_LOCAL_ENGINE=postgres
WAREHOUSE_BIGQUERY_PROJECT=TODO_PROJECT_ID
WAREHOUSE_BIGQUERY_LOCATION=US
WAREHOUSE_BIGQUERY_RAW_DATASET=commerce_raw
WAREHOUSE_BIGQUERY_STAGING_DATASET=commerce_staging
WAREHOUSE_BIGQUERY_MARTS_DATASET=commerce_marts
WAREHOUSE_BIGQUERY_PUBLIC_PROJECT=bigquery-public-data
WAREHOUSE_BIGQUERY_GA4_DATASET=ga4_obfuscated_sample_ecommerce
WAREHOUSE_BIGQUERY_THELOOK_DATASET=thelook_ecommerce
DBT_TARGET=dev
# Kestra
KESTRA_NAMESPACE=omnichannel.platform.dev
KESTRA_HOST=http://localhost:8080
# Spark
SPARK_MASTER_URL=local[*]
SPARK_APP_NAME=omnichannel-platform
# GCP / Terraform / dbt prod target
GCP_PROJECT_ID=TODO_PROJECT_ID
GCP_REGION=TODO_REGION
GCP_TERRAFORM_STATE_BUCKET=TODO_TF_STATE_BUCKET
GOOGLE_APPLICATION_CREDENTIALS=TODO_PATH_TO_SERVICE_ACCOUNT_JSON
GCP_SERVICE_ACCOUNT_JSON=TODO_INLINE_JSON_FOR_DBT_OR_CI
# Dashboard
DASHBOARD_PORT=8501
DASHBOARD_WAREHOUSE_SCHEMA=staging
DASHBOARD_RAW_SCHEMA=raw
API_BASE_URL=
# Cloud Run / dashboard image deployment
ARTIFACT_REPOSITORY_ID=omnichannel-platform
DASHBOARD_SERVICE_NAME=omnichannel-dashboard
DASHBOARD_CONTAINER_IMAGE=us-central1-docker.pkg.dev/TODO_PROJECT_ID/omnichannel-platform/dashboard:latest
This monolithic structure suffers from fatal architectural flaws: 1. Zero Domain Separation: Infrastructure parameters, catalog settings, and database configurations are thrown into one flat key-value namespace. 2. No Type Safety: Ports, partitions, memory strings, and boolean flags are parsed as raw strings, shifting casting responsibilities to application code. 3. No Validation: Typos in configuration keys fail silently during runtime environment lookup, causing delayed and opaque errors. 4. Security Risks: Public port parameters and non-sensitive setups sit right next to passwords and service account keys, forcing the entire file to be treated as high-risk and gitignored. 5. DRY Violated: Docker Compose, Python modules, and Shell utilities must each parse the file independently, duplicating key lookup logic.
Our Solution: Structured, Domain-Split YAML Configuration
To solve this, we centralize our non-sensitive configurations into hierarchical YAML files categorized by domain within config/ (e.g., config/ports.yml), parsed strictly in one location inside our library.
Decision Drivers
- Single Source of Truth (SSOT): Centralize every public parameter so that it is never duplicated.
- Maintainability & Readability: Enable comments, self-documentation, and hierarchical structure.
- Type Enforced Boundaries: Enforce explicit type casting and defaults immediately at load time.
- Strict Security Boundaries: Keep public configurations in source control (YAML) and sensitive secrets isolated in gitignored
.envfiles. - No Duplicate Shell Parsers: Enable external consumers (Docker) to query values without writing custom YAML parsers in Bash.
Alternatives
- Alternative A: Monolithic
.envFile — Store all environment variables and configuration options globally in a single.envfile at the root. - Pros: Native integration with standard Docker and shell utilities.
- Cons: Zero type safety, no hierarchical groupings, no comments, high security exposure (mixing public config with secrets).
- Alternative B: Domain-Specific
.envFiles — Split configurations into multiple files (e.g.,spark.env,ports.env). - Pros: Better logical separation.
- Cons: Still lacks structural hierarchy, no native validation, cluttered file listing.
- Alternative C: Centralized YAML + Typed Loader (SELECTED) — Store public configurations in structured YAML, parse them through an immutable Python dataclass loader, and use a
justfilewrapper to inject parameters to external consumers. - Pros: Perfect structure, comments supported, explicit type enforcement, complete separation of secrets, fully DRY.
- Cons: Small execution overhead (1-2s) when running Python from
justcommands.
Decision Framework
| Model / Option | Readability & Organization (Weight: 35%) | Type Safety & Validation (Weight: 30%) | Security Isolation (Weight: 20%) | Integration DRYness (Weight: 15%) | Total Score | Decision |
|---|---|---|---|---|---|---|
| Alternative C (Selected) | 10/10 | 10/10 | 10/10 | 9/10 | 9.85 | ✅ Selected |
| Alternative B | 6/10 | 1/10 | 7/10 | 5/10 | 4.55 | Rejected |
| Alternative A | 2/10 | 0/10 | 2/10 | 6/10 | 2.00 | Rejected |
Decision
We will adopt Alternative C (Centralized YAML + Typed Loader) to address the configuration scaling and DRY-violation issues. This involves using domain-specific YAML files in the config/ directory, loaded via typed, frozen PortsConfig dataclasses, with a justfile bridge to dynamically populate ports for Docker Compose. This decision establishes the absolute single source of truth for the local Lakehouse.
High-Level Architecture
graph TD
subgraph config/ [Centralized Config Directory]
A[ports.yml]
end
subgraph entsoe_pipeline [Python Package]
B[paths.py] -->|CONFIG_DIR| C[config_loader.py]
C -->|PortsConfig dataclass| D[Python Pipeline Runtime]
end
subgraph task_runner [Justfile Orchestrator]
E[justfile] -->|Calls config_loader| D
E -->|Exports env vars| F[Docker Compose]
end
A -->|Parsed by| C
Related Requirements
Functional Requirements
- FR-1: System must centralize infrastructure port mappings to prevent port conflicts across the local environment.
- FR-2: The platform must dynamically adapt to port overrides without rebuilding Docker images.
Non-Functional Requirements
- NFR-1 (Security): Configuration files must not contain sensitive tokens or database passwords to prevent accidental commits to public repositories.
- NFR-2 (Maintainability): The configuration parsing logic must be fully isolated to a single entry-point module (
config_loader.py). - NFR-3 (Extensibility): New configuration domains must be addable by extending the YAML structure without altering existing runtime loaders.
Performance Requirements
- PR-1: Configuration load times must be sub-second to prevent performance degradation during task initialization.
- PR-2: Memory footprint of configuration instances must be negligible using immutable dataclass representations.
Integration Requirements
- IR-1: Docker Compose must dynamically inherit port configuration from the centralized YAML values at container startup.
- IR-2: Pre-commit hooks must validate YAML syntax to ensure only syntactically sound configuration is committed.
Related Decisions
- ADR-002 (Local Environment Configuration): Development environment setup with library-first tooling using
uvfor environment execution.
Design
Architecture Overview
sequenceDiagram
participant Config as config/ports.yml
participant Loader as config_loader.py
participant Just as justfile
participant Docker as docker-compose.yml
Note over Config: Single Source of Truth
Just->>Loader: uv run python -c "from entsoe_pipeline..."
Loader->>Config: Parse config file
Config-->>Loader: raw dict
Loader-->>Loader: Enforce types (int)
Loader-->>Just: port stdout (e.g. "8333")
Just->>Docker: Export SEAWEEDFS_S3_PORT=8333
Docker->>Docker: Substitute port mappings dynamically
Implementation Details
In src/entsoe_pipeline/config/config_loader.py:
"""Configuration loader module for the ENTSO-E data pipeline.
This module provides type-safe, immutable configuration objects by parsing
infrastructure YAML files and loading environment parameters.
"""
from dataclasses import dataclass
import yaml
from entsoe_pipeline import CONFIG_DIR
@dataclass(frozen=True)
class PortsConfig:
"""Immutable infrastructure ports configuration.
Attributes:
seaweedfs_s3 (int): Port for SeaweedFS S3-compatible API.
iceberg_catalog (int): Port for Apache Iceberg REST Catalog.
"""
seaweedfs_s3: int
iceberg_catalog: int
@classmethod
def from_yaml(cls) -> "PortsConfig":
"""Loads and parses the ports configuration from ports.yml.
Returns:
PortsConfig: A type-safe configuration object.
Raises:
FileNotFoundError: If ports.yml is missing.
yaml.YAMLError: If ports.yml contains invalid syntax.
"""
ports_file = CONFIG_DIR / "ports.yml"
with open(ports_file, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
ports_data = data.get("ports", {})
# Enforce type casting and fallback values
return cls(
seaweedfs_s3=int(ports_data.get("seaweedfs_s3", 8333)),
iceberg_catalog=int(ports_data.get("iceberg_catalog", 8181)),
)
In src/entsoe_pipeline/config/paths.py:
import os
from pathlib import Path
def find_project_root() -> Path:
"""Locates the project root directory deterministically."""
if env_root := os.getenv("PROJECT_ROOT"):
return Path(env_root)
current_file = Path(__file__).resolve()
for parent in current_file.parents:
if (parent / ".project_root").exists():
return parent
return current_file.parents[3]
PROJECT_ROOT = find_project_root()
DATA_DIR = PROJECT_ROOT / ".data"
TESTS_DIR = PROJECT_ROOT / "tests"
ADR_DIR = PROJECT_ROOT / "docs" / "adr"
CONFIG_DIR = PROJECT_ROOT / "config"
In src/entsoe_pipeline/__init__.py:
"""ENTSO-E Data Pipeline library."""
from entsoe_pipeline.config.paths import (
ADR_DIR,
DATA_DIR,
PROJECT_ROOT,
TESTS_DIR,
CONFIG_DIR,
)
__all__ = ["ADR_DIR", "DATA_DIR", "PROJECT_ROOT", "TESTS_DIR", "CONFIG_DIR"]
Configuration
In config/ports.yml:
# Infrastructure ports configuration for local Lakehouse
ports:
seaweedfs_s3: 8333
iceberg_catalog: 8181
In justfile:
seaweedfs_s3_port := `uv run python -c "from entsoe_pipeline.config.config_loader import PortsConfig; print(PortsConfig.from_yaml().seaweedfs_s3)"`
iceberg_catalog_port := `uv run python -c "from entsoe_pipeline.config.config_loader import PortsConfig; print(PortsConfig.from_yaml().iceberg_catalog)"`
# Dynamically export variables so Docker Compose automatically inherits them
export SEAWEEDFS_S3_PORT := seaweedfs_s3_port
export ICEBERG_CATALOG_PORT := iceberg_catalog_port
# Start the local Lakehouse services in the background
up:
@echo "[JUST] Launching SeaweedFS"
@echo "[JUST] S3 port: {{seaweedfs_s3_port}}"
@echo "[JUST] Iceberg port: {{iceberg_catalog_port}}"
docker compose -f docker/docker-compose.yml up -d
@echo "[JUST] SeaweedFS is up and running"
# Stop and tear down all infrastructure containers
down:
@echo "[JUST] Stopping SeaweedFS"
docker compose -f docker/docker-compose.yml down
@echo "[JUST] SeaweedFS stopped"
In docker/docker-compose.yml:
services:
seaweedfs:
image: chrislusf/seaweedfs:latest
container_name: seaweedfs
restart: unless-stopped
ports:
- "9333:9333" # HTTP master
- "19333:19333" # gRPC master
- "8080:8080" # HTTP Volumes
- "8888:8888" # HTTP Filer
- "18888:18888" # gRPC Filer
- "${SEAWEEDFS_S3_PORT:-8333}:8333" # S3 API
- "${ICEBERG_CATALOG_PORT:-8181}:8181" # Iceberg REST Catalog
volumes:
- /mnt/seaweed_data:/data
command: 'server -dir=/data -s3'
Testing
In tests/test_config.py:
import pytest
from unittest.mock import mock_open, patch
from entsoe_pipeline.config.config_loader import PortsConfig
def test_ports_config_loading():
"""Verify that configuration values are correctly parsed and cast to integers."""
mock_yaml = """
ports:
seaweedfs_s3: 9000
iceberg_catalog: 9001
"""
with patch("builtins.open", mock_open(read_data=mock_yaml)):
config = PortsConfig.from_yaml()
assert config.seaweedfs_s3 == 9000
assert config.iceberg_catalog == 9001
def test_ports_config_fallbacks():
"""Verify that missing keys default safely to fallback ports."""
mock_yaml = "ports: {}"
with patch("builtins.open", mock_open(read_data=mock_yaml)):
config = PortsConfig.from_yaml()
assert config.seaweedfs_s3 == 8333
assert config.iceberg_catalog == 8181
Consequences
Positive Outcomes
- Absolute Single Source of Truth: Changing a port configuration in
ports.ymlseamlessly propagates to both python pipeline environments and docker-compose containers. - Type Enforced Safety: Ports are parsed and validated as integers instantly at load time.
- Robust Domain Isolation: Clear separation between git-monitored environments (YAML) and sensitive execution secrets (untracked
.env). - Completely DRY Tooling: Reuses the exact same Python dataclass parser via
justfileinstead of duplicated shell utilities.
Negative Consequences / Trade-offs
- Small Startup Overhead: Each
justfileinvocation performsuv run pythonto fetch the variables, causing a 1-second delay. - Environment Dependency: Requires the Python environment with dependencies (e.g.
PyYAML) to be resolved locally to start containers via thejustfile.
Ongoing Maintenance & Considerations
- Ensure YAML syntax correctness via linting tools in CI to prevent parse failures.
- Track additional infrastructure configuration parameters by expanding domain YAML files (e.g.,
spark.yml,duckdb.yml) rather than.envkeys.
Dependencies
- Infrastructure:
Docker >= 20.10,just >= 1.20 - Data Frameworks:
PyYAML >= 6.0 - Removed: Any legacy shell-script parser or
.envtracking for infrastructure ports.
References
- YAML serialization specifications
- Twelve-Factor App guidelines for configuration management
- Just Command Runner specifications
Changelog
- v1.0 (2026-05-19): Initial accepted version.
Title
Centralized Path Single Source of Truth (SSOT) Configuration and AST Quality Gates
Description
Establish a centralized path management architecture using config/paths.yml as the Single Source of Truth (SSOT), dynamically populated as Path objects in Python via paths.py relative to a deterministic .project_root anchor, and strictly enforced across the entire codebase using AST quality gate unit tests.
Context
The Problem: Hardcoded Relative Paths and Runtime Layout Fragility
A common anti-pattern in complex data engineering projects is to resolve paths in-place inside individual source modules. Typically, this looks like:
# Fragile, hardcoded path resolution in application logic
from pathlib import Path
DATA_DIR = Path(__file__).resolve().parents[2] / "data" / "raw"
This ad-hoc approach has severe architectural flaws:
1. Runtime Portability Failure: When executing code under different runtimes (such as a local shell, inside Docker containers, within Apache Spark clusters, or triggered by an orchestrator like Airflow), the working directory and file hierarchy change. Hardcoded relative traversals (via multi-level parent directory traversal) break immediately when the execution context changes.
2. Refactoring Nightmare: If the project directory layout changes (e.g., relocating fms_metadata/ or .data/ folders), developers must manually search and update every single file that traverses or constructs paths.
3. Untestable Code: Mocking filesystem operations or redirecting paths to isolated temporary test directories is extremely difficult when paths are constructed locally inside target functions.
4. Configuration Drift: Having multiple files declare where configurations or data directories reside inevitably leads to silent drift, where different jobs expect data in slightly different locations.
Our Solution: Centralized declarative path registry
To eliminate this fragility, we establish a centralized, declarative paths configuration registry. A base root directory is located deterministically using a file-based anchor (.project_root) and environment variable fallbacks, while all subdirectories and configuration files are registered in a single YAML file and exported dynamically.
Decision Drivers
- Single Source of Truth (SSOT): Ensure every directory and file path in the workspace is defined in exactly one place.
- Portability & Environment Agnosticism: Guarantee path resolution works identically in development, CI, testing, and production (Docker/Spark).
- Refactoring Ease: Allow updating the entire directory hierarchy layout by changing a single line of configuration.
- Strict Compliance: Enforce that developers cannot bypass the SSOT by writing raw paths or in-place resolutions in the codebase.
Alternatives
- Alternative A: Hardcoded Relative Path Resolution — Resolve paths in-place relative to module files (
__file__) or current working directory (os.getcwd()). - Pros: Simple to write initially, zero dependencies.
- Cons: Extremely fragile, breaks under Spark/Docker environments, impossible to easily refactor directories, hard to mock during tests.
- Alternative B: Pure Environment Variable Configuration — Inject all paths as environment variables via shell scripts or
.envfiles. - Pros: Decoupled from codebase structure.
- Cons: Clutters shell scripts, no static analysis validation, prone to typos, high configuration boilerplate.
- Alternative C: Centralized Declarative Path Registry + AST Quality Gates (SELECTED) — Declare paths in
config/paths.yml, load them dynamically in Python, locate the project root using a.project_rootanchor, and audit compliance with automated tests. - Pros: Single configuration file, absolute layout portability, clean type-safe
Pathimports, zero boilerplate, and 100% automated enforcement. - Cons: Dynamic module attribute generation requires fallback hooks (
__getattr__) to satisfy static type checkers and IDE autocompletion tools.
Decision Framework
| Model / Option | Portability & Robustness (Weight: 35%) | Maintainability & DRY (Weight: 30%) | Code Cleanliness (Weight: 20%) | Enforcement & Safety (Weight: 15%) | Total Score | Decision |
|---|---|---|---|---|---|---|
| Alternative C (Selected) | 10/10 | 10/10 | 9/10 | 10/10 | 9.80 | ✅ Selected |
| Alternative B | 7/10 | 6/10 | 5/10 | 3/10 | 5.70 | Rejected |
| Alternative A | 1/10 | 2/10 | 4/10 | 0/10 | 1.75 | Rejected |
Decision
We will adopt Alternative C (Centralized Declarative Path Registry + AST Quality Gates).
- All paths are defined relative to the project root in config/paths.yml.
- The project root is resolved deterministically by project_root.py using the .project_root marker file or a PROJECT_ROOT environment override.
- The paths.py library interface loads the YAML file and dynamically populates its module namespace.
- We implement test_paths_ssot_quality_gate.py to statically audit all files, ensuring no raw path constructions or standard library resolutions bypass this registry.
High-Level Architecture
Related Requirements
Functional Requirements
- FR-1: The system must locate the project root directory regardless of whether it is run from a terminal shell, a test runner, or inside containerized workloads.
- FR-2: The application must expose all configured paths as standard
pathlib.Pathobjects to ensure cross-platform compatibility (Windows/Linux/macOS).
Non-Functional Requirements
- NFR-1 (Maintainability): Modifying any directory path must only require changing the configuration in
config/paths.yml. - NFR-2 (Testability): All path variables must be mockable/overrideable during unit and integration test executions.
- NFR-3 (Strict Verification): The workspace layout registry compliance must be validated automatically as part of the test suite.
Performance Requirements
- PR-1: Path resolution must execute in sub-millisecond times, caching references to avoid redundant lookup processes.
Integration Requirements
- IR-1: The registry configuration must run transparently across CLI, local unit test runner sessions, Docker mounts, and PySpark workers.
Related Decisions
- ADR-001 (Centralized YAML Configuration): Establishes the standard configuration layout patterns under
config/andconfig_env/.
Design
Architecture Overview
Implementation Details
In src/entsoe_pipeline/config/core/project_root.py:
"""Project root lookup implementation core module."""
from __future__ import annotations
import os
from pathlib import Path
def find_project_root() -> Path:
"""Locates the project root directory deterministically.
Resolves the root folder using a three-tier fallback strategy:
1. Prioritizes the 'PROJECT_ROOT' environment variable (useful in Docker/Airflow).
2. Searches parent directories upward for the '.project_root' anchor file.
3. Falls back to static parent calculation relative to this file's path.
"""
if env_root := os.getenv("PROJECT_ROOT"):
return Path(env_root)
current_file = Path(__file__).resolve()
for parent in current_file.parents:
if (parent / ".project_root").exists():
return parent
return current_file.parents[4]
In src/entsoe_pipeline/config/paths.py:
import typing
from entsoe_pipeline.config.core.project_root import find_project_root
from entsoe_pipeline.config.config_loader import load_paths_config
PROJECT_ROOT = find_project_root()
PATHS_YML = PROJECT_ROOT / "config" / "paths.yml"
# Load paths dynamically from the SSOT configuration file
_paths_data = load_paths_config(PROJECT_ROOT)
# Populate module namespace dynamically to establish SSOT exports
for _key, _rel_val in _paths_data.items():
globals()[_key] = PROJECT_ROOT / _rel_val
# Expose constants for package-wide utilization
__all__ = ["PROJECT_ROOT"]
__all__.extend(list(_paths_data.keys()))
def __getattr__(name: str) -> typing.Any:
"""Allow dynamic attributes for static type checkers.
This magic method ensures that IDEs, static linters, and type checkers (like
mypy/pyright) recognize the dynamically populated path constants as valid
module-level attributes, preventing false-positive lint warnings.
"""
if name in __all__:
return globals().get(name)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
Configuration
In config/paths.yml:
# ENTSO-E Metadata Pipeline Path Configurations
# This file is the Single Source of Truth (SSOT) for all workspace paths.
# All path values are declared relative to the PROJECT_ROOT directory.
DATA_DIR: ".data"
TESTS_DIR: "tests"
ADR_DIR: "docs/adr"
API_LIMITS_YML: "config/entsoe_api_limits.yml"
CONFIG_DIR: "config_env"
CONFIG_EXAMPLE_DIR: "config_env_example"
ENV_FILE: ".env"
FMS_METADATA_DIR: "fms_metadata"
OVERVIEW_YML: "fms_metadata/overview.yml"
OVERVIEW_TREE_YML: "fms_metadata/overview_tree.yml"
EXPORT_LOG_YML: "fms_metadata/export_log.yml"
LANDING_BUCKET_SCHEMA_YML: "config/entsoe_fms_folder_schema.yml"
Testing
To ensure absolute adherence to our paths convention, we perform two layers of validation:
- Convention Compliance (Quality Gate): We statically parse all
.pyfiles using regular expressions to audit code patterns. Any script attempting to resolve parent hierarchies usingPath(__file__), checkingos.getcwd(), or instantiating hardcoded path strings pointing to workspace folder targets will fail this check. - Symmetry and Parity Tests: We ensure that every uppercase constant dynamically exported by
paths.pymatches the declarative list registered inpaths.yml, preventing undocumented path constants.
In tests/test_paths_ssot_quality_gate.py:
import re
from pathlib import Path
import pytest
from entsoe_pipeline import SRC_DIR, TESTS_DIR
EXEMPT_FILES = {"paths.py", "project_root.py", "test_paths.py", "test_paths_ssot_quality_gate.py"}
def get_python_files():
# Recursively yield all source and test modules, excluding EXEMPT_FILES
return [p for p in SRC_DIR.rglob("*.py") if p.name not in EXEMPT_FILES]
@pytest.mark.parametrize("file_path", get_python_files())
def test_no_raw_paths_or_parent_traversal(file_path: Path):
"""Quality gate verifying that modules import path constants from paths.py."""
content = file_path.read_text(encoding="utf-8")
violations = []
# Audit for in-place relative traverses
if re.search(r"Path\(\s*__file__\s*\)", content):
violations.append("Found reference to Path(__file__). Centralize path in paths.py")
if re.search(r"PROJECT_ROOT\s*/", content):
violations.append("Direct raw path concatenation using PROJECT_ROOT. Export from paths.py")
if re.search(r'Path\(\s*["\'](?:\.env|\.data|config_env)["\']\s*\)', content):
violations.append("Direct instantiation of hardcoded repository paths. Import from paths.py")
assert not violations, f"Path violations found in {file_path.name}:\n" + "\n".join(violations)
Consequences
Positive Outcomes
- Absolute Portability: Running tests or pipeline runs works natively across local laptops, staging setups, Docker environments, and PySpark clusters without path failures.
- Easy Maintenance: Moving a directory layout requires changing exactly one YAML line, rather than editing dozens of hardcoded paths.
- Improved Testability: Unit tests can safely patch or redirect specific directories (like the landing bucket schema path) by simply altering
paths.LANDING_BUCKET_SCHEMA_YMLwithout complex filesystem mocking. - No Hardcoding Drift: Ensures new developers cannot introduce ad-hoc layouts, enforcing clean development hygiene.
Negative Consequences / Trade-offs
- Static Analyzer Workarounds: Dynamic namespace injection using
globals()bypasses standard static code compilation checks. To ensure IDE autocompletions and linters still recognize the constants, we had to introduce__getattr__fallback hooks and declare them in package exports. - Overhead in Setup: Adding a new directory path requires adding it to both
paths.ymland updating the coverage test dictionary intest_paths.py.
Ongoing Maintenance & Considerations
- New paths must always be defined in
config/paths.ymland mirrored in theEXPECTED_PATHSmapping insidetests/test_paths.py. - Any exceptions/exemptions to the quality gate (such as utility scripts or path-specific test suites) must be explicitly listed in the
EXEMPT_FILESset intests/test_paths_ssot_quality_gate.py.
Dependencies
- Infrastructure:
.project_rootempty anchor file located at the project root directory. - Libraries:
pathlib(Python Standard Library),ruamel.yaml/yaml(Parser).
References
- ADR-001: Centralized YAML Configuration
- Python PEP 562 -- Module getattr and dir
- Twelve-Factor App guidelines on configuration and path portability
Changelog
- v1.0 (2026-06-15): Initial accepted version detailing centralized dynamic YAML paths configuration and AST quality gates.
Title
Centralized Config Loader Public Facade and Typed Core Model Separation
Description
Establish a type-safe, modular configuration loading architecture utilizing src/entsoe_pipeline/config/config_loader.py as a pure public facade (API) that delegates all logic to specialized, immutable dataclasses defined within the core/ package, resolving configurations dynamically relative to our path registry without hardcoding directory locations.
Context
The Problem: Monolithic configuration loading and path assumptions
A metadata-driven pipeline depends on numerous configuration inputs:
- Infrastructure ports (ports.yml)
- Storage buckets (bucket.yml)
- Target AWS regions (region.yml)
- System hostnames (hosts.yml)
- Keycloak credentials (enviroment.yml)
- Rate limiting specifications (entsoe_api_limits.yml)
- Data classifications rules (entsoe-classifier.yml)
If each ingest job or helper script parses these YAML files directly using custom code and hardcoded relative paths, it results in:
1. High Fragility: Moving folders or modifying schema structure breaks multiple files silently.
2. Boilerplate Duplication: Identical loading, validation, and parsing logic is copied across jobs.
3. No Cache Optimization: Reading configuration files repeatedly from disk during run loops hurts pipeline performance.
4. Poor IDE Experience: Accessing configuration properties as dynamic dictionaries (such as config.ports.s3_compatible) prevents IDE autocompletion, static type checking, and compile-time error detection.
Our Solution: Separated core parsers and public loader API
We decouple the configuration schema definitions from the retrieval API. Domain schemas are declared as typed, frozen Python dataclasses in config/core/. A clean, cached public loader interface (config_loader.py) acts as the exclusive gateway for all consumer code.
Decision Drivers
- Facade pattern encapsulation: Hide configuration parsing and YAML mapping logic behind a clean, unified public interface.
- Strict type validation: Enforce types, mandatory parameters, and sensible fallbacks on module load.
- Portability & Agnosticism: Never hardcode directory targets (like
fms_metadata/,config_env/, orconfig/) in code; resolve them exclusively using thepaths.pyregistry (SSOT). - Automated quality gates: Prevent configuration structure or metadata descriptions from drifting out of sync.
Alternatives
- Alternative A: In-Place Dictionary Loading — Load raw YAML files dynamically into dict objects whenever a script needs them.
- Pros: Quick to write, zero class boilerplate.
- Cons: No type safety, no property autocompletion, duplicative path traversals, prone to key errors.
- Alternative B: Monolithic Config Class — Declare a single massive configuration class that loads all YAML directories at once.
- Pros: Simpler file layout (one class).
- Cons: Poor separation of concerns, violates single-responsibility principle, harder to test individual sub-systems.
- Alternative C: Facade Loader + Typed Core Package (SELECTED) — Isolate domains into separate dataclass files within
config/core/and expose them exclusively through cached accessors inconfig_loader.py. - Pros: Maximum modularity, clean public API, cached memory footprints, 100% type safety.
- Cons: Multiple file creation overhead.
Decision Framework
| Model / Option | Modularity & DRY (Weight: 35%) | Type Safety & Cache (Weight: 30%) | Test Isolation (Weight: 20%) | Code Cleanliness (Weight: 15%) | Total Score | Decision |
|---|---|---|---|---|---|---|
| Alternative C (Selected) | 10/10 | 10/10 | 10/10 | 9/10 | 9.85 | ✅ Selected |
| Alternative B | 6/10 | 8/10 | 6/10 | 7/10 | 6.75 | Rejected |
| Alternative A | 2/10 | 1/10 | 2/10 | 3/10 | 1.85 | Rejected |
Decision
We will adopt Alternative C (Facade Loader + Typed Core Package).
- Public config API is defined exclusively in src/entsoe_pipeline/config/config_loader.py.
- Facade delegates all parsing and schemas to individual files in src/entsoe_pipeline/config/core/.
- No paths are hardcoded; all configuration directories (config_env/, config/, fms_metadata/) are resolved via paths.py.
High-Level Architecture
graph TD
A["config_env/ & config/ YAML Files"] -->|Parsed by classmethods| B["Core Modules config/core/*"]
B -->|Instantiates typed models| C["config_loader.py Facade"]
C -->|Exposes cached singletons| D["Pipeline Ingest Jobs / Tests"]
E["paths.py SSOT"] -->|Resolves file locations| B
Related Requirements
Functional Requirements
- FR-1: All configuration options must be accessible via type-safe attributes rather than string dictionary lookups.
- FR-2: The configuration system must cache loaded configs in memory to avoid repetitive disk I/O.
Non-Functional Requirements
- NFR-1 (Maintainability): The facade script must remain thin and free of direct class schemas, acting purely as an import bridge.
- NFR-2 (Robustness): All parameters must fall back to safe default ports or options if configuration files are partially empty.
Performance Requirements
- PR-1: Loading configurations from cache must execute in sub-microsecond timescales to avoid degrading performance during tight iteration loops.
Integration Requirements
- IR-1: Configuration parameters must expose native representations compatible with environment configurations, Docker daemon network constraints, and AWS service specifications.
Related Decisions
- ADR-001 (Centralized YAML Configuration): Establishes the folder structure split between
config/andconfig_env/. - ADR-002 (Centralized Path SSOT Configuration): Resolves configuration paths dynamically using
paths.pyrelative to.project_root.
Design
Architecture Overview
sequenceDiagram
participant Job as Ingestion Job / Script
participant Facade as config_loader.py
participant Core as core/pipeline.py
participant Model as core/ports.py
participant Paths as paths.py
Job->>Facade: get_ports_config()
Note over Facade: Cache check (LRU)
Facade->>Paths: Resolve CONFIG_DIR path
Paths-->>Facade: Path("/workspace/config_env")
Facade->>Model: PortsConfig._from_yaml()
Note over Model: Read yml & validate types
Model-->>Facade: PortsConfig instance
Facade-->>Job: PortsConfig (cached)
Implementation Details
The core logic and type schemas reside in the following modules:
1. src/entsoe_pipeline/config/core/buckets.py: Declares BucketsConfig for S3 bucket namespace mapping.
2. src/entsoe_pipeline/config/core/classifier.py: Declares ClassifierConfig for FMS catalog folder indexing rules.
3. src/entsoe_pipeline/config/core/hosts.py: Declares HostsConfig mapping hostnames of internal containers.
4. src/entsoe_pipeline/config/core/limits.py: Declares RateLimitsConfig defining throttle constraints.
5. src/entsoe_pipeline/config/core/ports.py: Declares PortsConfig defining TCP network mapping targets.
6. src/entsoe_pipeline/config/core/region.py: Declares RegionConfig for geographical cloud target identification.
7. src/entsoe_pipeline/config/core/pipeline.py: Composes all sub-configs into a single PipelineConfig master model.
Sample Facade Accessor - in src/entsoe_pipeline/config/config_loader.py:
from functools import cache
from entsoe_pipeline.config.core import PortsConfig
@cache
def get_ports_config() -> PortsConfig:
"""Loads and returns the cached PortsConfig singleton."""
return get_config().ports
Configuration
Example configuration structure inside config_env/my_entsoe_domains.yml:
# Schema mapping for our active domains configuration
active_mode: "Default"
config_name: "statistical_must_haves"
Testing
We enforce the config loader design and metadata completeness using three specialized testing strategies:
-
Facade Integrity Tests (
tests/test_config_loader.py): Verifies thatconfig_loader.pycontains zero direct class definitions (all class models must be imported from thecore/package). It also asserts the correctness of defaults, caches, and missing file error raises. -
Metadata Specification Audit (
tests/test_config_metadata.py): Validates that all example/template configuration files (e.g., inconfig_env_example/) contain ametadatasection documenting every parameter. It enforces that: - Descriptions are present and comprehensive (>= 150 characters).
- Valid data types (
string,integer,object) are declared. -
Default values match their declared types.
-
Folder Schema Mirror Audit (
tests/test_config_mirror.py): Enforces that the keys and structures inconfig_env_example/match the developers' local configuration directoryconfig_env/exactly. This prevents developers from introducing new properties locally without documenting them in the public template files.
Consequences
Positive Outcomes
- Clean API Contract: Ingestion scripts simply call
get_config()and read attributes without worrying about relative paths, filesystem parsing, or dict keys. - Robust Verification: Developers are forced to fully document and match template configuration changes, preventing broken onboarding setups.
- Optimized Execution: Repetitive config queries return cached objects in sub-microseconds without hitting the filesystem.
Negative Consequences / Trade-offs
- File Proliferation: Adding a single configuration parameter domain requires creating a YAML configuration file, a core class model file, a facade wrapper function, and updating mirror tests.
Ongoing Maintenance & Considerations
- Any new configurations added to the repository must be accompanied by corresponding template mappings in
config_env_example/to satisfy the schema mirror audit tests. - When updating schema properties, their types must be mirrored inside
config/core/package schemas.
Dependencies
- Libraries:
pydanticor typed dataclasses withfunctools.cachefor runtime representation and memoization. - Configurations: The path system defined in
paths.py(ADR-002) is required to locate the configuration folder targets.
References
- Twelve-Factor App guidelines on configuration segregation
- Design Patterns: Facade Structural Pattern
Changelog
- v1.0 (2026-06-15): Initial accepted version detailing centralized dynamic config loader facade design.
Title
Ephemeral Landing Zone with Event-Driven Staging Integration
Description
Establish an ephemeral landing zone architecture for the ENTSO-E data pipeline, where raw CSV files are immediately processed into Apache Iceberg tables in the Staging (Silver) layer and subsequently purged from S3, keeping only the landing registry as the historic metadata passport of raw ingestions.
Context
- Storage Optimization: Raw CSV files downloaded from the ENTSO-E FMS API are large (often 50+ MB per month/endpoint). Storing them permanently in S3 is redundant and results in massive storage costs and cluster disk usage.
- Silver Layer Primacy: Downstream analysis, Spark processing, and querying are executed solely on the Silver (Staging) Iceberg tables, which store the data in highly compressed Parquet format.
- Auditability ("Data Passport"): Even if raw CSV files are deleted, the pipeline requires an audit trail to track what files were downloaded, when they were ingested, and their hashes (to prevent duplicate downloads unless the remote file changes). The local
landing_registry.jsonfile serves as this historical metadata database.
Decision Drivers
- Reduce object storage consumption in SeaweedFS.
- Maintain transactional consistency and historic traceability.
- Automate downstream pipeline triggering on file arrivals.
Alternatives
- Alternative A (Persistent Raw Files): Retain all raw CSV files in S3. This consumes large amounts of storage and offers no architectural benefit.
- Alternative B (No Registry): Purge raw files without updating a registry, losing all synchronization history and duplicate prevention logic.
Decision Framework
| Model / Option | Solution Leverage (Weight: 30%) | Application Value (Weight: 40%) | Maintenance (Weight: 30%) | Total Score | Decision |
|---|---|---|---|---|---|
| Ephemeral Landing (Selected) | 9/10 | 9/10 | 8/10 | 8.7 | ✅ Selected |
| Persistent Raw Files | 4/10 | 5/10 | 6/10 | 5.0 | Rejected |
Decision
We will adopt an Ephemeral Landing with Event-Driven Staging pattern. Raw CSV files in the landing-zone bucket act as a temporary buffer and are purged from S3 upon successful load into Iceberg, with metadata persisted to the landing_registry.json.
High-Level Architecture
graph TD
FTP[ENTSO-E FMS FTP Server] -->|1. Sync new/modified CSVs| S3_Landing[(S3 landing-zone bucket)]
S3_Landing -->|2. Trigger ETL Job| Spark[Spark Transformer Job]
Registry[(landing_registry.json)] -->|Lookup hashes/metadata| Spark
Schemas[(iseberg_schemas_registry.json)] -->|Read target schema & cast| Spark
Spark -->|3. Append/Merge to Iceberg| Iceberg[(S3 lakehouse bucket / Iceberg Silver)]
Spark -->|4. Delete Raw CSV| S3_Landing
Spark -->|5. Mark status as 'purged'| Registry
Related Requirements
Functional Requirements
- FR-1: The pipeline must ingest new and modified files idempotently using xxHash comparison.
- FR-2: The pipeline must purge raw files after successful database load to minimize storage.
Non-Functional Requirements
- NFR-1: S3 Landing Zone storage utilization must remain minimal.
- NFR-2:
landing_registry.jsonmust act as a complete, read-only historic passport of all ingested files.
Performance Requirements
- PR-1: Spark read and write operations must complete within 2 minutes per monthly CSV chunk.
- PR-2: Memory footprints for the Spark execution environment must remain under 4GB local executor bounds.
Integration Requirements
- IR-1: The sync engine must dynamically write schemas using PySpark REST catalog connector endpoints.
- IR-2: Kestra must trigger the sync and transform flows based on file state events.
Related Decisions
- ADR-002 (Centralized Path SSOT): All registry and schema JSON files conform to paths resolved dynamically by
paths.yml.
Design
Architecture Overview
The system divides ingestion into a lightweight FTP sync step and a transactional Spark-based Silver transformation step.
Implementation Details
- FTP Ingestion: Sync script downloads files to
landing-zone/. - Spark Transformation: Spark reads the raw CSV, applies formatting and overrides from
schema_overrides.yml, writes to Iceberg, and executes S3 deletion.
Configuration
The purge behavior is managed via config_env/lakehouse.yml:
lakehouse:
purge_raw_after_transform: true
Testing
- Unit Tests: Verify schema parsing and overrides application in isolation.
- Integration Tests: Execute mock sync runs, verify table appends in Iceberg Catalog, and check that files are purged from the landing zone mock bucket.
Consequences
Positive Outcomes
- Significant storage savings.
- Declarative schemas.
- Clean database state.
Negative Consequences / Trade-offs
- Re-ingesting raw files requires re-downloading them from FMS.
- Spark REST Catalog must remain active for all transform jobs.
Ongoing Maintenance & Considerations
- Schema overrides must be updated in
schema_overrides.ymlif column names change.
Dependencies
- PySpark 4.1.1 runtime bundle.
- SeaweedFS S3-compatible endpoints.
References
- ENTSO-E FMS Integration Specification.
- Apache Iceberg Spark Integration Documentation.
Changelog
- v1.0 (2026-06-29): Initial proposed draft.
Title
Pull Request-Driven Git Commit Workflow for Ingestion Pipeline Development
Description
Adopt a Pull Request (PR)-driven development workflow where detailed context, rationale, and directory-level modifications are documented comprehensively in a single consolidated PR description rather than verbose, granular git commits.
Context
During the bootstrapping phase of this educational ENTSO-E pipeline, the overhead of writing atomic, detailed git commits for every minor file change is high. Verbose commit messaging consumes substantial developer time and drains AI API token budgets during pair programming sessions. Since the repository is in an early stage (zero stars, single-developer environment), optimizing development velocity and time-to-market is the primary objective. We need a compromise that preserves historical documentation without slowing down active development.
Decision Drivers
- Developer Velocity: Speed up time-to-market by reducing friction in the commit lifecycle.
- API Token Conservation: Prevent excessive token consumption by avoiding frequent prompts to generate atomic commit messages.
- Centralized Documentation: Maintain clean, comprehensive change summaries at the PR level.
Alternatives
- Option A (Atomic Verbose Commits): Write highly detailed, multi-line messages for every single commit. (High overhead, slows down development).
- Option B (Empty/Unstructured Commits): Write vague or empty commit messages without any centralized documentation. (Creates high technical debt, untraceable changes).
- Option C (PR-Driven Consolidation): Use pragmatic commit messages on feature branches, but compile a highly detailed description of all changes in the final PR when merging to main. (Selected: optimizes velocity while keeping documentation high).
Decision Framework
| Model / Option | Developer Velocity (Weight: 40%) | Token Conservation (Weight: 30%) | Granular Traceability (Weight: 30%) | Total Score | Decision |
|---|---|---|---|---|---|
| Option C (PR-Driven) | 9/10 (3.6) | 9/10 (2.7) | 6/10 (1.8) | 8.1 | ✅ Selected |
| Option A (Atomic) | 4/10 (1.6) | 3/10 (0.9) | 9/10 (2.7) | 5.2 | Rejected |
| Option B (Empty) | 8/10 (3.2) | 9/10 (2.7) | 2/10 (0.6) | 6.5 | Rejected |
Decision
We will adopt Option C (PR-Driven Consolidation). Developers will work on feature branches (following standard branch workflows). Feature branch commits can use concise, practical summaries. When the feature branch is complete, a single, comprehensive Pull Request will be created. The PR description will contain a complete summary of all changes, files modified, and test verification logs. When merging, the merge commit will reference this PR (e.g. feat: staging schemas generator (#12)), establishing a clean history.
High-Level Architecture
Feature Branch (Concise Commits)
--> [Commit 1: Setup schema caster]
--> [Commit 2: Setup table identifier]
--> [Create Pull Request #12: Consolidated Staging Transformation Plan & Logs]
--> [Merge PR into main -> Single Merge Commit referencing PR #12]
Related Requirements
Functional Requirements
- FR-1: Developers must be able to commit changes quickly without writing multi-line explanations for every commit.
- FR-2: The merge process must link to a PR that hosts the comprehensive description.
Non-Functional Requirements
- NFR-1 (Maintainability): The main branch history must remain clean, highlighting high-level feature merges.
- NFR-2 (Usability): Code reviews must contain exhaustive checklists of changes.
Performance Requirements
- PR-1: Commit submission overhead must not exceed 5 seconds per commit.
Integration Requirements
- IR-1: The GitHub/GitLab repository must enforce pull request usage for main branch merges.
Related Decisions
- ADR-001 (Centralized YAML Configuration): Ensures configurations are decoupled from scripts, reducing the size of code changes.
- ADR-002 (Centralized Path SSOT): Minimizes file paths changes across codebases, reducing git conflicts during merges.
Design
Architecture Overview
Git branch policy will restrict direct pushes to the main branch. All developments must go through feature branches and pull requests.
Implementation Details
A standard PR template will be established in the repository to guide PR creation:
## Summary of Changes
- Refactored `spark/core/iseberg_schema_generator.py` to parse files internally.
- Created `jobs/staging/lakehouse/generate_iceberg_schemas.py` job executor.
## Verification
- Run `just generate-schemas` (output: successful logs).
- Run `uv run pytest tests/` (all passed).
Configuration
Branch protection rules will require a reviewed PR prior to merging to main.
Testing
Verification of this workflow will be managed during pull request reviews. The automated pre-commit hooks (like validate_adrs.py and validate_proposals.py) will run in CI to verify document structures.
Consequences
Positive Outcomes
- Token Efficiency: Minimizes AI context/token usage for commit generation.
- Velocity: Speeds up code delivery.
- Documentation Quality: Detailed PR descriptions are easier to read than multiple scattered git commits.
Negative Consequences / Trade-offs
- Granular git blame logs on lines of code will point to individual concise commits, requiring developers to trace back to the merge PR to understand specific details.
Ongoing Maintenance & Considerations
- Ensure PRs are squash-merged or merge commits are properly referenced so history remains readable.
Dependencies
- Infrastructure: Git version control, GitHub repository policies.
References
- GitHub Flow Guide - Pull request-based development workflows.
Changelog
- v1.0 (2026-06-29): Initial version defining PR-driven commit workflow.
Title
Isolation of IOP Test Environment and FMS Credentials
Description
Define the architectural boundary between the ENTSO-E Integration Test Platform (IOP) and the Production (PROD) platforms, limiting the storage of IOP test data to temporary validation buffers and isolating developer/test credentials.
Context
- Analytical Quality of Data: The ENTSO-E IOP platform is a sandbox environment for data providers to test their publishing pipelines. The files published in IOP contain mock data, synthetic values, and fragmented timelines. Loading this data into the staging (Silver) lakehouse layer creates data pollution and distorts analytical results.
- API and FMS Rate Limits: The ENTSO-E servers enforce strict rate limiting (400 requests/minute). Running automated pipeline tests or local developer runs against the PROD servers risk exceeding these limits, causing service blocks for production jobs.
- CI/CD Validation Safety: Developers require a sandbox environment to test synchronization scripts, parser logic, and Iceberg table writes in CI/CD pipelines. This must run without impacting production storage or credentials.
Decision Drivers
- Isolate production data from synthetic test data.
- Prevent API rate-limit starvation on production FMS servers.
- Enable robust integration testing in CI/CD pipelines.
Alternatives
- Alternative A (Single Unified Environment): Mix IOP and PROD datasets in the same S3 landing bucket and write them into unified database tables. This results in heavy data pollution.
- Alternative B (Dual Staging Buckets): Deploy separate S3 lakehouse buckets for test and production runs, and load IOP data permanently. This increases cloud costs for zero analytical value.
- Alternative C (IOP as CI/CD Test Sandbox): Store IOP files under an isolated directory structure in the landing zone bucket, skip loading IOP data to the persistent staging layer, and use IOP credentials exclusively for validation tests.
Decision Framework
| Model / Option | Solution Leverage (Weight: 30%) | Application Value (Weight: 40%) | Maintenance (Weight: 30%) | Total Score | Decision |
|---|---|---|---|---|---|
| IOP as CI/CD Test Sandbox (Selected) | 9/10 | 9/10 | 8/10 | 8.7 | ✅ Selected |
| Dual Staging Buckets | 5/10 | 4/10 | 6/10 | 4.9 | Rejected |
| Single Unified Environment | 2/10 | 2/10 | 5/10 | 2.9 | Rejected |
Decision
We will adopt the IOP as CI/CD Test Sandbox approach.
1. The landing zone bucket retains both iop/ and prod/ folder paths for temporary raw downloads to support local debugging.
2. Staging jobs will only load files from the prod/ raw folders into the persistent lakehouse catalog.
3. Automated integration tests and CI/CD pipelines will execute using IOP credentials to test full-cycle FMS downloads and staging transformations safely.
High-Level Architecture
graph TD
PROD_FMS[PROD FMS Server] -->|Production Sync| Landing_Prod[s3://landing-zone/prod]
IOP_FMS[IOP Test FMS Server] -->|Test/CI Ingestion| Landing_Iop[s3://landing-zone/iop]
Landing_Prod -->|ETL Staging Job| Iceberg[(S3 lakehouse/Silver)]
Landing_Iop -.->|CI/CD Verification Job| Temp_Iceberg[(Temporary Test Tables)]
Temp_Iceberg -->|Clean Up| Trash((Purged))
Related Requirements
Functional Requirements
- FR-1: The sync job must dynamically point to either the IOP or PROD FMS servers based on the active environment configuration.
- FR-2: Staging transformation pipelines must filter out IOP data from production tables.
Non-Functional Requirements
- NFR-1: (Data Quality) Staging lakehouse tables must remain free of synthetic or test records.
- NFR-2: (Security) Production credentials must never be exposed or used during automated test suites.
Performance Requirements
- PR-1: CI/CD runs executing against the IOP server must respect the 400 requests/minute API limit.
Integration Requirements
- IR-1: The test runner must resolve active credentials dynamically using environment resolvers.
Related Decisions
- ADR-001 (Centralized YAML Config): Environment URLs and credential keys are resolved from env configurations.
- ADR-004 (Ephemeral Landing Zone): Raw CSV files are cleared from the S3 landing-zone bucket after staging.
Design
Architecture Overview
The system isolates environment endpoints using config_env/enviroment.yml, which defines base URLs and token servers for IOP and PROD.
Implementation Details
In src/entsoe_pipeline/config/env_resolver.py:
def resolve_active_environment() -> str:
# Dynamically reads the active environment (IOP or PROD)
In jobs/staging/landing/prepare_landing_ingestion.py:
active_env = resolve_active_environment()
# executes crawler using active_env configurations
Configuration
In config_env/enviroment.yml:
active_environment: "PROD"
environments:
IOP:
base_url: "https://fms.tp-iop.entsoe.eu/"
PROD:
base_url: "https://fms.tp.entsoe.eu/"
Testing
In tests/test_io_operations.py:
def test_iop_sync_workflow(monkeypatch):
"""Verify that setting the environment to IOP successfully routes requests."""
monkeypatch.setenv("ENV", "IOP")
# Assertions for IOP FMS host routing
Consequences
Positive Outcomes
- Complete isolation of production analytics from synthetic test data.
- Elimination of rate-limiting conflicts on production FMS servers during developer activity.
- Secure, reproducible CI/CD integration testing.
Negative Consequences / Trade-offs
- Requires managing two sets of client credentials (IOP and PROD).
- Synthetic data in IOP may occasionally drift from production formats, requiring format adaptations in tests.
Ongoing Maintenance & Considerations
- Monitor ENTSO-E platform change announcements for schema changes in both environments.
- Rotate FMS client tokens on a scheduled basis.
Dependencies
- Infrastructure: S3-compatible SeaweedFS, ENTSO-E FMS API endpoints.
References
- ENTSO-E Transparency Platform Guide
- ADR-001: Centralized Configuration
- ADR-004: Ephemeral Landing Zone
Changelog
- v1.0 (2026-06-29): Initial draft defining the IOP environment isolation strategy.
Title
Separation of Global FMS Metadata Refresher from Active Landing Ingestion
Description
Define the architectural boundary between crawling the complete physical FMS metadata catalog (a heavy, long-running discovery task) and preparing the landing zone schemas for specific active data domains (a lightweight production task).
Context
- Recursive Crawling Correctness: A flat FMS crawler (non-recursive) failed to discover files nested deep within monthly/daily subfolders in active domains (like
OfferedTransferCapacitiesContinuousEvolution_11.1_r3), leading to incorrect physical size reports (e.g. 0.0 MB) and data loss. Implementing recursive listing solved this but increased the number of API calls significantly. - API Request Overhead: Crawling the entire FMS remote file system recursively (across test, prod, and legacy publication folders) requires thousands of requests. Due to client-side rate limits (95 requests/minute), a complete recursive crawl takes between 1.5 to 3 hours.
- Operational Decoupling:
- Exploration & Schema Mapping: Developers and analysts need a complete, updated physical map of FMS (
overview.yml,overview_tree.yml) to understand available datasets, schemas, and historical data weights. This is an exploration task. - Production Data Ingestion: The hourly ingestion pipeline (
ingest_my_entsoe_domains.py) only needs a preflight schema contract (entsoe_fms_folder_schema.yml) of the subset of active domains configured for the current environment. This is a runtime execution task. - Running the entire global crawler before every data ingestion job is computationally wasteful, rate-limit starving, and introduces a huge bottleneck to production SLAs.
Decision Drivers
- Reduce the execution time and API call overhead of the staging landing preparation job.
- Retain complete and accurate recursive metadata discovery for FMS platform exploration.
- Prevent API rate-limit starvation on production FMS servers.
- Maintain a clean separation between development/exploration artifacts and active production schema registries.
Alternatives
- Alternative A (Keep Unified Heavy Job): Run the complete recursive crawl on every preparation run. This guarantees the latest schema but increases prep runtime to 2+ hours, making hourly ingestion impossible.
- Alternative B (Selective Ingestion In-Place): Keep the logic in the same job, but add environment flags to skip legacy crawls. This leaves the code coupled and still relatively slow if even one heavy domain (like transmission capacities) is active.
- Alternative C (Decouple Global Discovery from Active Prep): Extract the global FMS crawling process into an independent orchestrator script (
jobs/refresh_fms_metadata.py) run weekly/monthly. Restrict the active ETL preparation job (prepare_landing_ingestion.py) to process only the active domains configuration checklist (my_entsoe_domains.yml).
Decision Framework
| Model / Option | Solution Leverage (Weight: 30%) | Application Value (Weight: 40%) | Maintenance (Weight: 30%) | Total Score | Decision |
|---|---|---|---|---|---|
| Decouple Global Discovery (Selected) | 9/10 | 9/10 | 9/10 | 9.0 | ✅ Selected |
| Selective Ingestion In-Place | 6/10 | 5/10 | 6/10 | 5.6 | Rejected |
| Keep Unified Heavy Job | 2/10 | 3/10 | 5/10 | 3.3 | Rejected |
Decision
We will adopt the Decouple Global Discovery from Active Prep approach.
- Extract Global Refresher: Extract the full-tree recursive crawling of FMS directories into a dedicated, heavy-duty utility job at
jobs/refresh_fms_metadata.py. This job will run asynchronously (e.g. once a week/month) to update the global exploration catalogs (fms_metadata/overview.yml,fms_metadata/overview_tree.yml) in Git. - Restrict Active Prep: The standard orchestrator
prepare_landing_ingestion.pywill only crawl and process the specific active domains selected inmy_entsoe_domains.yml. It will no longer execute the globalingest_fms_metadatastep. - Isolate Active Metadata Storage: Active metadata slices required to construct the S3 landing schema contract will be written to a dedicated
.data/fms_metadata/directory, leaving the rootfms_metadata/directory reserved exclusively for global developer/analyst catalogs.
High-Level Architecture
graph TD
subgraph Heavy Discovery (Periodic / Monthly)
Crawler_Heavy[jobs/refresh_fms_metadata.py] -->|Recursive Crawl| FMS_All[All FMS Directories]
FMS_All -->|Compile & Commit| Git_Doc[fms_metadata/overview_tree.yml]
end
subgraph Active Production Ingestion (Hourly)
Checklist[.data/my_entsoe_domains.yml] -->|Reads Active Domains| Prep_Job[prepare_landing_ingestion.py]
Prep_Job -->|Lightweight Crawl| FMS_Active[Active Domains Only]
FMS_Active -->|Compile Schema| Schema_Registry[.data/fms_metadata/schema_contract.yml]
Schema_Registry -->|Validation & Execution| Ingest_Job[ingest_my_entsoe_domains.py]
end
Related Requirements
Functional Requirements
- FR-1:
refresh_fms_metadata.pymust support full recursive FMS crawler loops for all active and legacy publication paths. - FR-2:
prepare_landing_ingestion.pymust query the active domains checklist to build a restricted list of target directories, bypassing any inactive branches.
Non-Functional Requirements
- NFR-1: (Maintainability) Production crawler configurations must reside in declarative configuration folders, separate from Python ETL logic.
- NFR-2: (Decoupling) Production ingestion runs must not depend on or modify the global documentation catalogs in
fms_metadata/.
Performance Requirements
- PR-1: The hourly landing preparation job must run in under 2 minutes when running standard active domain checklists.
- PR-2: The FMS API client-side rate limits must be strictly throttled to 95 requests per minute to prevent temporary platform IP bans.
Integration Requirements
- IR-1: The active preparation job must resolve active credentials dynamically using environment resolvers.
- IR-2: Schema compilation results must comply with the central S3 landing bucket schema registry specifications.
Related Decisions
- ADR-001 (Centralized YAML Config): Environment URLs and credential keys are resolved from env configurations.
- ADR-002 (Centralized Path SSOT): Path constants for active FMS metadata storage in
.data/will be defined centrally. - ADR-006 (IOP Sandbox): Limits the exposure of test sandbox credentials.
Design
Architecture Overview
By isolating the heavy global refresh task from the active ingestion preparation, we protect the production data flows from API rate limits and minimize latency. The active pipeline relies on a clean, trimmed schema contract generated dynamically from the specific domains that are currently being ingested.
Implementation Details
In jobs/staging/landing/prepare_landing_ingestion.py:
def main() -> None:
active_env = resolve_active_environment()
with RunsLogger(job_name="prepare_landing_ingestion", environment=active_env):
generate_my_entsoe_domains()
# Resolve active_folders_by_domain and active_legacy_folders from MY_ENTSOE_DOMAINS_YML
# Run crawlers only for active folders pointing to ACTIVE_PHYSICAL_CATALOG_DIR
for dom, folders in active_folders_by_domain.items():
ingest_func(env=active_env, folders=folders, catalog_dir=ACTIVE_PHYSICAL_CATALOG_DIR)
# Compile active tree and active sizes
ingest_overview_tree_metadata(ACTIVE_PHYSICAL_CATALOG_DIR, ACTIVE_OVERVIEW_TREE_YML)
build_landing_bucket_schema(ACTIVE_OVERVIEW_TREE_YML, LANDING_BUCKET_SCHEMA_YML)
ingest_all_catalog_sizes(ACTIVE_PHYSICAL_CATALOG_DIR, ACTIVE_SIZES_DIR)
Configuration
In config/paths.yml:
# Active FMS metadata directory paths inside the workspace
ACTIVE_FMS_METADATA_DIR: ".data/fms_metadata"
ACTIVE_PHYSICAL_CATALOG_DIR: ".data/fms_metadata/physical_catalog"
ACTIVE_SIZES_DIR: ".data/fms_metadata/sizes"
ACTIVE_OVERVIEW_TREE_YML: ".data/fms_metadata/overview_tree.yml"
Testing
Verification Commands:
# Path verification and dynamic schema loader test
uv run python -m pytest tests/test_paths.py
Consequences
Positive Outcomes
- Reduces landing preparation job latency from hours to seconds for targeted ingestion.
- Eliminates redundant API calls and rate-limiting issues on the production ENTSO-E servers.
- Preserves full-catalog historical documentation mapping for team onboarding.
- Ensures staging lakehouse pipelines do not ingest unneeded domain telemetry.
Negative Consequences / Trade-offs
- Running the global crawler requires manual triggering or separate orchestration scheduler DAGs.
- Active schemas are calculated incrementally, which might lead to outdated schema files if a new domain is activated without running preparation first.
Ongoing Maintenance & Considerations
- Audit the rate-limit limits periodically to optimize FMS client concurrency.
- Schedule the
refresh_fms_metadata.pyjob as a cron task in the orchestration environment.
Dependencies
- Infrastructure:
SeaweedFS,Docker. - Data Frameworks:
PySpark >= 4.0.0,requests >= 2.31.0.
References
- ENTSO-E Transparency Platform File Directory User Guide - Official guide to FMS layout
- ADR-002: Centralized Path SSOT - Regarding data path configuration lifecycle management
Changelog
- v1.0 (2026-06-29): Initial proposed version describing the separation of FMS metadata refresh from active landing ingestion.
Title
Migration of FMS Metadata Catalog from YAML to PostgreSQL
Description
Migrate the storage format of the FMS physical metadata catalog from local flat YAML files to a structured PostgreSQL database. This migration resolves memory exhaustion OOM crashes, enforces transactional ACID guarantees, ensures idempotency, and enables low-latency hierarchical directory tree lookups for downstream BI/UI interfaces.
Context
In the previous metadata architecture (ADR-007), we separated the global metadata catalog refresh process from the landing zone ingestion. The metadata crawler recursively searched the remote FMS FTP servers and compiled metadata details (file sizes, counts, update times, hashes) into localized YAML catalog files under fms_metadata/physical_catalog.
As the crawlers achieved deep recursive crawls across active and historical archives:
- The IOP catalog reached 74K+ files (63 GB original data size).
- The PROD catalog exceeded 218K+ files (9.4 TB original data size), with the Transmission and Balancing domains being particularly massive.
- The Python crawling process accumulated the entire directory structure in memory as a nested dictionary, attempting to serialize it to a YAML catalog via yaml.dump at the end of the crawl.
- On June 30, the PROD Balancing crawl process consumed 12.4 GB of RAM and was terminated by the host system's earlyoom daemon, demonstrating that the sheer size of the Balancing domain makes YAML storage impossible.
- Parsing the resulting 92 MB Transmission.yml file back into Python memory takes 60–90 seconds, causing severe performance bottlenecks.
- YAML provides no transactional (ACID) safety. If a crawl is interrupted, the catalog is left in a partially written or corrupted state, forcing a complete 3-hour re-scan from scratch.
- Realizing resume-capable incremental scanning with YAML is impractical because checking folder state requires parsing massive files on every startup.
Decision Drivers
- Reliability & OOM Prevention: Ingestion memory utilization must remain low and constant (O(1)) instead of scaling linearly with the file count.
- Idempotency & Partial Updates: Support upserting metadata entries on-the-fly in small batches, resuming aborted crawls without duplicate data.
- Query Performance & Hierarchy: Enable low-latency recursive directory tree traversal and filtering directly in database queries.
- BI Compatibility: Provide a standard SQL connection interface for downstream BI dashboards (e.g. Apache Superset, Metabase).
Alternatives
- A: Streaming YAML / Binary Serialization (e.g.
yaml.safe_dump_allor msgpack) - Pros: Keeps metadata file-based, requiring no database services.
- Cons: No query interface. BI/UI dashboards would still have to parse large files. Lack of concurrent write protection.
- B: Embedded Databases (SQLite / DuckDB)
- Pros: Embedded, zero infrastructure, supports SQL.
- Cons: Single-writer locking limitations. If the IOP and PROD metadata crawler threads attempt concurrent writes, SQLite/DuckDB will throw database lock errors.
- C: PostgreSQL
- Pros: Reuses the existing PostgreSQL database stack running for Kestra, fully ACID compliant, supports high write concurrency, and supports efficient recursive CTEs for tree traversals.
- Cons: Requires keeping a database service running (already covered by docker-compose).
Decision Framework
| Model / Option | Solution Leverage (Weight: 30%) | Concurrency & ACID (Weight: 35%) | BI Integration (Weight: 20%) | Infrastructure Complexity (Weight: 15%) | Total Score | Decision |
|---|---|---|---|---|---|---|
| PostgreSQL | 9 | 10 | 10 | 8 | 9.45 | ✅ Selected |
| SQLite | 8 | 5 | 6 | 9 | 6.70 | Rejected |
| DuckDB | 8 | 4 | 5 | 9 | 6.15 | Rejected |
Decision
We will adopt PostgreSQL to host the FMS physical metadata catalog, replacing the flat YAML files under fms_metadata/physical_catalog.
We will implement this using the SQLModel library (which integrates Pydantic schemas and SQLAlchemy models) and the psycopg database driver. We will reuse the existing PostgreSQL instance (kestra_postgres) by creating a dedicated database named entsoe_metadata.
This decision supersedes the file-based catalog layout specified in ADR-007.
High-Level Architecture
graph TD
A[FMS FTP Server] -->|Page-by-page Crawl| B[Python Metadata Crawler]
B -->|Batch Upserts| C[(PostgreSQL: entsoe_metadata)]
C -->|Recursive CTE Query| D[UI Directory Tree]
C -->|Standard SQL SQL| E[BI Dashboards]
Related Requirements
Functional Requirements
- FR-1: The system must record detailed stats (names, sizes, update dates, hashes) of every FMS file.
- FR-2: The database must model the parent-child relationships of the FMS directory tree.
Non-Functional Requirements
- NFR-1 (ACID Reliability): Ingestion updates must be transactional to prevent dirty metadata states.
- NFR-2 (Idempotency): The system must use upserts (
ON CONFLICT) to ensure safe reruns.
Performance Requirements
- PR-1 (RAM Efficiency): Ingestion RAM utilization must not exceed 200 MB, even when crawling millions of files.
Integration Requirements
- IR-1: The database must support concurrent connections from multiple crawlers and UI backends.
Related Decisions
- ADR-007 (Separation of Metadata Refresh from Ingestion): Replaces the YAML file serialization stage of the metadata refresh job with direct database writing.
Design
Architecture Overview
We define two tables: fms_folders and fms_files in a parent-child relationship.
Implementation Details
# Model definitions using SQLModel
from typing import Optional
from sqlmodel import SQLModel, Field, Relationship
class FMSFolder(SQLModel, table=True):
__tablename__ = "fms_folders"
id: Optional[int] = Field(default=None, primary_key=True)
environment: str = Field(index=True) # 'iop' or 'prod'
domain: str = Field(index=True) # 'Balancing', 'Transmission', etc.
folder_path: str = Field(index=True)
parent_id: Optional[int] = Field(default=None, foreign_key="fms_folders.id")
item_count: int = 0
original_bytes: int = 0
compressed_bytes: int = 0
crawled_at: Optional[datetime] = Field(default=None) # UTC timestamp of last crawl
class FMSFile(SQLModel, table=True):
__tablename__ = "fms_files"
file_id: str = Field(primary_key=True) # UUID from FMS
folder_id: int = Field(foreign_key="fms_folders.id")
name: str = Field(index=True)
original_bytes: int
compressed_bytes: int
last_updated: str
xxhash: str
Configuration
We configure database credentials via our env-specific YAML configs using paths defined in the config loader.
database:
url: "postgresql://kestra:password@localhost:5432/entsoe_metadata"
Testing
We will implement database interaction tests using pytest and a running test PostgreSQL container or mock sessions.
# Skeleton test logic
import pytest
from sqlmodel import Session, SQLModel, create_engine
def test_metadata_idempotent_upsert():
# Arrange: Setup SQLite memory engine for fast unit tests
engine = create_engine("sqlite:///:memory:")
SQLModel.metadata.create_all(engine)
# Act: Perform double insert of same metadata
# Assert: Count is 1, size is updated
pass
Consequences
Positive Outcomes
- Permanently resolves memory OOM issues by writing data in small streaming batches.
- Avoids data corruption via SQL transaction boundaries (ACID).
- Drastically reduces query latency for dashboard reporting (from 90s to milliseconds).
- Enables efficient resume-capable incremental crawling (idempotency). By checking the
crawled_attimestamp in the database, the crawler can skip fresh folders in milliseconds. This is governed by theconfig/fms_crawler.ymlconfiguration (e.g.freshness.max_age_days: 3). Attempting this with flat YAML files was impossible due to the high parsing overhead. - Simplifies the ingestion code by eliminating multiple identical script entry points (dry refactoring).
Negative Consequences / Trade-offs
- Introduces database schema migration overhead (requires using Alembic or raw DDL migrations).
- Requires a running PostgreSQL instance (satisfied by our current docker setup).
Ongoing Maintenance & Considerations
- Monitor indexing performance on query paths (e.g. indices on
folder_pathandenvironment). - Clean up historical catalog entries if needed to keep the table sizes optimized.
Dependencies
- Infrastructure:
PostgreSQL >= 16 - Data Frameworks:
SQLModel >= 0.0.22,psycopg >= 3.1.0
References
- SQLModel Documentation - Integrated database model definitions
- PostgreSQL WITH RECURSIVE - Documentation on CTE hierarchical traversals
Changelog
- v1.1 (2026-07-01): Updated to document database-backed folder freshness tracking (
crawled_at) and thefms_crawler.ymlfreshness config enabling low-overhead resume-capable scans. - v1.0 (2026-06-30): Initial proposed draft migration from file-based YAML catalogs to PostgreSQL.
Title
FMS Metadata Catalog Matching and Contracts
Description
Enforce exact folder suffix matching (e.g. _r3) in user-level configuration templates as a strict Data Contract, while retaining prefix-based startswith() matching in the crawling discovery core to tolerate minor revisions seamlessly.
Context
In our metadata pipeline architecture, folders on the ENTSO-E FMS FTP server are physically revisioned with suffixes indicating major data schema or business logic updates (e.g. /TP_export/ActualTotalLoad_6.1.A_r3/). The logical names for these items in documentation and the Zendesk help page omit this suffix.
When designing user-level active domain checklists (like my_entsoe_default.yml), there is a temptation to abstract these suffixes away to make the configurations look cleaner. However, doing so leaks abstract definitions into physical execution boundaries.
If the pipeline automatically resolved and swallowed major revision changes (e.g. automatically matching a newly released _r4 directory via prefix matching), it would lead to Silent Data Corruption:
- Incompatible schema modifications or physical metric changes (e.g. units shifts, column splits) would be written directly into the S3 landing Bronze layer.
- Downstream tables using formats like Apache Iceberg might tolerate the schema changes due to physical schema evolution, but the underlying business aggregations would silently output corrupt or incomplete results.
- Additionally, prefix collisions exist (e.g. OfferedTransferCapacitiesContinuous_11.1_r3 vs OfferedTransferCapacitiesContinuousEvolution_11.1_r3), which makes loose matching unsafe.
Decision Drivers
- Data Quality & Reliability: Prevent Silent Data Corruption in analytical tables by failing fast on breaking schema revisions.
- Explicit Data Contracts: Ensure user configuration files explicitly state the target directory schema revision being consumed.
- Fail-Safe Integrity: Do not swallow major API/FTP breaking structure changes automatically.
- Collision Avoidance: Prevent prefix matching collisons on folders with similar root names.
Alternatives
- A: Dynamic Prefix Matching via startswith() in User Configs
- Pros: Simpler, cleaner configs without
_r3suffixes. - Cons: Silent data corruption if
_r4is released; namespace collisions between continuous and continuous evolution domains. - B: Strict Equality Mapping (==) Everywhere
- Pros: Fully strict contract.
- Cons: Extremely fragile. If ENTSO-E releases a minor patch folder like
_r3.1(which has the same schema but minor changes), the crawler will crash, requiring manual user intervention and config edits for non-breaking changes.
Decision Framework
| Model / Option | Solution Leverage (Weight: 25%) | Concurrency & ACID (Weight: 20%) | Data Reliability (Weight: 35%) | Infrastructure Complexity (Weight: 20%) | Total Score | Decision |
|---|---|---|---|---|---|---|
| Hybrid Contracts | 9 | 10 | 10 | 9 | 9.55 | ✅ Selected |
| Dynamic Matching | 8 | 9 | 4 | 9 | 7.20 | Rejected |
| Strict Equality | 7 | 9 | 9 | 9 | 8.50 | Rejected |
Decision
We will adopt the Hybrid Contracts pattern. We will explicitly declare the target physical folder suffixes (e.g. _r3) in both the Single Source of Truth classifier (entsoe_domains_classifier.yml) and the user-level configuration templates.
We will verify active folders by comparing the configured template values with the discovered FTP folders using strict string equality. Concurrently, the crawling classifier engine will use prefix-based startswith() matching to assign FTP directories to domains, enabling the pipeline to tolerate minor revision updates (e.g. _r3.1) dynamically while failing fast on major changes (e.g. _r4).
High-Level Architecture
graph TD
A[User Config: _r3] -->|Strict Equality Match| B[Overview Tree Catalog]
B -->|Crawl Filter| C[FMS FTP Server]
C -->|Discover _r3.1 Minor Patch| D[Crawler Core: startswith Match]
D -->|Classify & Ingest| E[(PostgreSQL: entsoe_metadata)]
C -->|Discover _r4 Major Release| F[Fail-Fast & Crash Ingestion]
Related Requirements
Functional Requirements
- FR-1: The system must restrict file downloading only to the exact revision directories specified in the active configuration.
- FR-2: The metadata crawler must successfully classify FMS directories into domains.
Non-Functional Requirements
- NFR-1: (Data Reliability) The pipeline must prevent silent data ingestion errors during upstream FMS schema updates.
- NFR-2: (Maintainability) All folder contracts must be centrally defined inside configuration files.
Performance Requirements
- PR-1: Configuration matching latency must not exceed 50 milliseconds.
Integration Requirements
- IR-1: The system must natively tolerate minor schema updates supported by the Apache Iceberg storage layout.
Related Decisions
- ADR-008 (Migration of FMS Metadata Catalog from YAML to PostgreSQL): The relational metadata tables will store the
folder_pathattribute containing the physical suffix. - ADR-007 (Separation of Metadata Refresh from Ingestion): The ingestion job matches selected folders against this database layout.
Design
Architecture Overview
The system divides matching into two phases:
1. Classifier Mapping: The entsoe_domains_classifier.yml acts as the SSOT mapping logical names to target physical FMS prefixes.
2. Custom Selection Matching: The configuration builder verifies custom selection lists against discovered folders.
Implementation Details
In src/entsoe_pipeline/fms_metadata/core/domain_classifier.py:
def classify_folder(folder_name: str) -> str:
"""Matches FMS directory name against prefix rules."""
config = get_classifier_config()
folder_name_lower = folder_name.lower()
for domain, items in config.domains.items():
for key, item in items.items():
fms_prefix = item.fms_name.lower()
if folder_name_lower.startswith(fms_prefix):
return domain
return config.fallback_domain
Configuration
In config/entsoe_domains_classifier.yml:
domains:
Load:
ActualTotalLoad_6.1.A:
name: "Actual Total Load [6.1.A]"
fms_name: ActualTotalLoad_6.1.A_r3
Testing
In tests/test_legacy_metadata_crawler.py:
import pytest
from entsoe_pipeline.fms_metadata.core.domain_classifier import classify_folder
def test_domain_classification_prefix_matching():
"""Verify that minor versions are classified correctly and major versions are ignored."""
# Correct R3 domain matching
assert classify_folder("ActualTotalLoad_6.1.A_r3") == "Load"
assert classify_folder("ActualTotalLoad_6.1.A_r3.1") == "Load"
# Non-matching R4 domain
assert classify_folder("ActualTotalLoad_6.1.A_r4") == "OtherMarketInformation"
Consequences
Positive Outcomes
- Prevents silent data corruption by blocking unvetted major schema transitions (
_r4). - Avoids pipeline crashes on harmless minor directory updates (
_r3.1) by relying on prefix startswith mapping. - Resolves all potential directory namespace collisions in similar domain prefixes.
- Simplifies configuration audits by keeping contracts explicit.
Negative Consequences / Trade-offs
- Requires manual developer update of classifier configurations when a major revision is rolled out.
- Increased configuration verbosity due to explicit suffixes in checklist templates.
Ongoing Maintenance & Considerations
- Audit the undocumented folders report to catch newly released FMS directory revisions.
- Update downstream silver ETL scripts before upgrading target configs to new major revisions.
Dependencies
- Infrastructure:
PostgreSQL - Data Frameworks:
Apache Iceberg
References
- Apache Iceberg Schema Evolution - How downstream tables manage added or modified attributes.
- Data Contracts in Modern Data Stacks - Best practices for API version boundary management.
Changelog
- v1.0 (2026-07-01): Initial accepted version documenting the Hybrid Contracts pattern for FMS suffixes.
Title
Dynamic Schema Inference Fallback for Unmapped Business Publications and Legacy Domains
Description
We will implement an automated schema inference fallback (using Spark's inferSchema) for business publications and legacy archives that lack structural definition details (i.e. columns: []) in the central entsoe_fms_schemas.yml configuration registry.
Context
The ENTSO-E File Management System (FMS) contains active and legacy directories that lack official HTML/Zendesk documentation from ENTSO-E (for example, older _r3 releases whose documentation was replaced by _r3.1, or legacy _r1 and _r2 archives).
Our crawler registers these publications with empty columns (columns: []). Under a strict schema enforcement model, attempting to create Iceberg tables for these publications results in database DDL compilation errors (empty column list syntax errors). Without a fallback, data engineers would have to manually inspect and code schemas for dozens of legacy endpoints, causing severe bottlenecks.
Decision Drivers
- Time-to-Market: Need to load historical archives without writing hundreds of manual overrides.
- Fail-Safe Processing: Isolated pipeline failures must not block unrelated data domains.
- Dynamic Schema Adaptability: The system must gracefully accept undocumented files while logging warnings.
Alternatives
- Option A (Strict Schema Enforcement): Manually map all missing columns in
entsoe_fms_schemas.yml. - Pros: High data quality guarantees, exact type casting.
- Cons: Extremely high manual maintenance overhead, blocks ingestion on new releases.
- Option B (Dynamic Schema Inference Fallback): Automatically trigger Spark schema auto-inference when
columns: []is encountered. - Pros: Low maintenance, high extensibility, immediate data load capabilities.
- Cons: Spark auto-inference may mistype float columns or timestamp formats, increasing downstream validation risk.
Decision Framework
| Model / Option | Time-to-Market (Weight: 40%) | Maintainability (Weight: 30%) | Schema Accuracy (Weight: 30%) | Total Score | Decision |
|---|---|---|---|---|---|
| Option B (Dynamic Inference Fallback) | 9/10 (3.6) | 9/10 (2.7) | 6/10 (1.8) | 8.1/10 | ✅ Selected |
| Option A (Strict Schema Enforcement) | 2/10 (0.8) | 4/10 (1.2) | 9/10 (2.7) | 4.7/10 | Rejected |
Decision
We will adopt Option B (Dynamic Schema Inference Fallback) to address unmapped legacy schemas. This involves using Spark's native inferSchema parameter dynamically when column list definitions are missing. This decision does not supersede any previous ADRs.
High-Level Architecture
We inject a fallback check within the Spark dataset loader. If a publication schema config does not contain defined columns, we bypass strict DDL generation and ask Spark to read the source file metadata headers dynamically.
[Landing Zone CSV] ──► [Empty Schema Checker] ──► (Yes) ──► [Spark inferSchema] ──► [Sanitize snake_case] ──► [Iceberg Table]
Related Requirements
Functional Requirements
- FR-1: Ingestion must process files with empty schemas in
entsoe_fms_schemas.yml. - FR-2: Target Iceberg tables must be created with dynamically inferred columns.
Non-Functional Requirements
- NFR-1: (Data Reliability) The fallback schema must prevent manual schema overrides where configuration is missing.
- NFR-2: (Maintainability) The config must be fully separated from PySpark script logic.
- NFR-3: (Extensibility) Schema migrations must occur without rebuilding the target tables.
Performance Requirements
- PR-1: Dynamic schema auto-inference must occur only once per table initialization to minimize metadata retrieval costs.
- PR-2: Memory footprint of Spark schema parsing must fit within standard executor allocations.
Integration Requirements
- IR-1: The system must natively connect to the target catalog (e.g. REST catalog).
- IR-2: Schema results must conform to Apache Iceberg metadata validation standards.
Related Decisions
- ADR-001 (Centralized YAML Configuration): Configuration schemas remain the Single Source of Truth (SSOT).
- ADR-009 (FMS Metadata Catalog Matching): Discovery uses early binding mapping.
Design
Architecture Overview
When Spark receives files to ingest:
1. It verifies the target table presence.
2. If absent and the contract schema has columns, it creates a table using mapped Spark columns.
3. If absent and the contract schema is empty, it reads the header row with inferSchema=True, maps column names to lower_snake_case, and creates the Iceberg table dynamically.
Implementation Details
In src/entsoe_pipeline/spark/landing_csv_reader.py:
# Target columns fallback to Spark schema inference if config list is empty
Configuration
In config/lakehouse_parquet_codec.yml:
No custom configurations are required; the fallback is dynamically resolved in Python code.
Testing
In tests/test_landing_csv_reader.py:
def test_schema_inference_fallback():
# Verify that a domain with columns: [] correctly infers schema from a raw CSV file
pass
Consequences
Positive Outcomes
- Reduces technical debt by eliminating manual schema evolution scripts.
- Unlocks loading of legacy archives immediately.
- Simplifies multi-environment deployments by abstracting catalog connections via environment-specific profiles.
Negative Consequences / Trade-offs
- Introduces strict schema enforcement, which may break downstream jobs if upstream changes are not backward compatible.
- Spark auto-inference may mistype float columns or timestamp formats.
Ongoing Maintenance & Considerations
- Audit catalog connection secrets rotation frequency.
- Track partition pruning efficiency in Spark UI to prevent full table scans.
Dependencies
- Infrastructure:
SeaweedFS,PostgreSQL. - Data Frameworks:
Apache Iceberg >= 1.11.0,PySpark >= 4.1.1. - Removed: None.
References
- Apache Iceberg Documentation - Detailed specifications for table maintenance.
- Parquet File Format Specifications - Technical insights into metadata.
Changelog
- v1.0 (2026-07-08): Initial version defining dynamic schema inference fallback rules.
Title
Decouple Embedded PySpark from the Kestra Orchestrator Container Using Spark Connect (gRPC)
Description
We will extract Spark execution (JVM, JARs, DataFrame processing) from the Kestra Worker container into a dedicated Spark Connect Server running on a separate compute node, reducing the orchestrator image from ~4.9 GB to ~300 MB.
Context
The current docker/Dockerfile installs pyspark with sql and connect extras, which pulls the full JVM runtime and all Iceberg/Hadoop JARs into the Kestra Worker container. This produces a 4.9 GB image that must be rebuilt and redeployed for any code change, even when only Python business logic is modified.
The production environment consists of three physical machines on a local network: Dell (storage/S3/Iceberg catalog), HP (16 GB RAM, compute worker), and ASUS (VPN/routing, Spark Master). A Spark standalone cluster already exists — Master on ASUS (spark://192.168.0.131:7077), Workers on HP (6 cores / 10 GB) and Dell (4 cores / 4 GB), all in ALIVE state. Despite this, Spark currently runs embedded inside the Kestra container — the same host that runs the orchestrator, ignoring the existing distributed cluster. This conflates two concerns:
- Orchestration (Kestra): schedules flows, manages task dependencies, executes lightweight Python crawling/logging.
- Heavy compute (Spark): reads CSVs from S3, performs MERGE INTO Iceberg tables, handles schema inference.
Embedding Spark inside Kestra forces the orchestrator to carry a full JVM runtime it never uses outside of specific staging jobs, inflating the image, slowing CI/CD pipelines, and coupling two independent operational concerns.
Spark Connect (gRPC-based client/server protocol, stable since Spark 3.4) allows PySpark to act as a thin client that serializes DataFrame operations over gRPC to a remote Spark Server. The client needs only the pyspark package with the connect extra (~50 MB), not the JVM or JARs. The existing Spark standalone cluster (Master + 2 Workers, 10 cores / 14 GB total) provides the execution backend — only a Connect Server needs to be started on the Master to bridge gRPC to the cluster.
Decision Drivers
- Image Size: 4.9 GB orchestrator image blocks fast iteration. A 300 MB image deploys ~16× faster.
- Separation of Concerns: Orchestration scheduling and heavy compute have different uptime, scaling, and debugging requirements.
- Existing Architecture: All Spark-consuming functions already accept
SparkSessionas a parameter — no business logic refactoring needed. - Existing Infrastructure: A Spark standalone cluster already runs: Master on ASUS, Workers on HP (6C/10GB) and Dell (4C/4GB). Embedded Spark ignores this investment.
- Operational Resilience: If the Spark Connect Server restarts, Kestra flows retry via the existing RunsLogger fault isolation (ADR-007). If Kestra restarts, Spark jobs on HP continue unaffected.
Alternatives
- Option A (Keep Embedded Spark — Status Quo): Continue shipping the full JVM inside the Kestra container.
- Pros: No new infrastructure, no gRPC dependency, simpler local development.
- Cons: 4.9 GB image, conflates compute and orchestration, ignores multi-machine topology.
- Option B (Spark Connect via Existing Cluster): Start a Spark Connect Server on the existing Master (ASUS), keep a thin PySpark client in Kestra. Workers on HP and Dell already running.
- Pros: 16× smaller orchestrator image, reuses existing cluster (no new containers), code already compatible.
- Cons: Spark Connect Server process to manage on ASUS, gRPC adds ~2-5 ms LAN overhead (negligible), local dev requires embedded fallback.
- Option C (Spark Job Server / Livy REST API): Submit Spark jobs via REST API instead of gRPC.
- Pros: REST is universally debuggable with curl, mature Livy ecosystem.
- Cons: Livy is semi-abandoned (last Apache release 2020), REST overhead for DataFrame serialization is higher than gRPC, requires job packaging as JARs/Python files rather than interactive sessions.
Decision Framework
| Model / Option | Image Size Reduction (Weight: 30%) | Code Change Surface (Weight: 35%) | Operational Fit (Weight: 35%) | Total Score | Decision |
|---|---|---|---|---|---|
| Option B (Spark Connect Split) | 10/10 (3.0) | 10/10 (3.5) | 9/10 (3.15) | 9.65/10 | ✅ Selected |
| Option A (Status Quo Embedded) | 0/10 (0.0) | 10/10 (3.5) | 5/10 (1.75) | 5.25/10 | Rejected |
| Option C (Livy REST API) | 9/10 (2.7) | 5/10 (1.75) | 7/10 (2.45) | 6.90/10 | Rejected |
Decision
We will adopt Option B (Spark Connect gRPC Client/Server Split) to decouple Spark execution from the Kestra orchestrator. The build_spark_session() factory in src/entsoe_pipeline/spark/spark_builder.py will detect the SPARK_CONNECT_URL environment variable. When set, it returns a thin gRPC-backed SparkSession pointing to the remote Spark Connect Server. When absent, it falls back to the existing embedded Spark for local development. A Spark Connect Server process is started on the existing Master node (ASUS), bridging gRPC to the already-running standalone cluster (spark://192.168.0.131:7077). Workers on HP (6C/10GB) and Dell (4C/4GB) provide execution. The Connect Server is configured with the same Iceberg/S3 connector JARs and catalog settings currently defined in spark_builder.py. No new Docker containers are required — the cluster infrastructure already exists. This decision does not supersede any previous ADRs.
High-Level Architecture
The existing Spark standalone cluster is reused as-is. A Spark Connect Server is started on the Master node (ASUS) as a gRPC bridge — the thin client in Kestra connects to it, and the Connect Server delegates execution to the existing Workers on HP and Dell.
┌─ Kestra Container (Dell) ─┐ ┌─ ASUS (Spark Master + Connect) ─┐
│ │ │ │
│ Kestra Worker │gRPC │ Spark Connect Server :15002 │
│ ├─ Python crawlers │─────►│ └─► Spark Master :7077 │
│ └─ spark_builder.py │ │ ├─► HP Worker (6C/10GB) │
│ .remote(sc://asus...) │◄─────│ └─► Dell Worker (4C/4GB) │
│ │gRPC │ │
│ Image: ~300 MB │ │ (cluster already exists) │
└────────────────────────────┘ └──────────────────────────────────┘
The gRPC boundary is transparent to all existing business logic. Functions like read_landing_csv_dataset(), merge_dataframe_into_table(), and add_csv_to_iceberg_table() receive a SparkSession parameter and operate identically regardless of whether it is local or remote.
Related Requirements
Functional Requirements
- FR-1: Kestra flows must execute Spark staging jobs (CSV ingestion, Iceberg MERGE) without a local JVM in the orchestrator container.
- FR-2: Embedded Spark must remain available for local development when
SPARK_CONNECT_URLis unset. - FR-3: The Spark Connect Server must be configured with the same Iceberg REST catalog, S3A endpoint, and JAR packages as the current embedded session.
Non-Functional Requirements
- NFR-1: (Separation of Concerns) Orchestrator image must contain only Python business logic and lightweight dependencies — no JVM, no Spark JARs.
- NFR-2: (Backward Compatibility) All existing Python code using
SparkSessionas a parameter must work without modification. - NFR-3: (Operational Independence) Kestra container restart must not interrupt running Spark jobs on HP. Spark Connect Server restart must be handled by Kestra's existing retry/fault isolation (ADR-007).
Performance Requirements
- PR-1: gRPC serialization overhead between Dell and HP must not exceed 10 ms per DataFrame operation (LAN: ~1 ms ping, overhead dominated by serialization).
- PR-2: Spark Workers provide 14 GB combined executor memory (HP: 10 GB, Dell: 4 GB). The existing cluster configuration is sufficient.
Integration Requirements
- IR-1: Spark Connect Server on ASUS:15002 must be reachable from the Kestra container via LAN (
192.168.0.131). - IR-2: All cluster nodes must have network access to SeaweedFS S3 (Dell:8333) and Iceberg REST catalog (Dell:8181) — already configured in the existing cluster.
Related Decisions
- ADR-004 (Ephemeral Landing Zone & Event-Driven Staging): The staging ingestion flow that triggers Spark jobs remains unchanged — only the execution location moves.
- ADR-007 (Separation of Metadata Refresh from Ingestion): The existing fault isolation pattern in
RunsLoggerhandles Spark Connect Server unavailability via retry and DLQ logging. - ADR-010 (Dynamic Schema Inference Fallback): The schema inference code in
landing_csv_reader.pyoperates on the same DataFrame API, transparently serialized over gRPC.
Design
Architecture Overview
The change is confined to a single module: spark_builder.py. All downstream consumers of SparkSession are parameterized and require no modification.
graph LR
subgraph "Kestra Container (~300 MB)"
J[ingest_landing_csv_to_lakehouse.py]
B[spark_builder.py]
end
subgraph "ASUS (Spark Master + Connect)"
S[Spark Connect Server :15002]
M[Spark Master :7077]
HW[HP Worker 6C/10GB]
DW[Dell Worker 4C/4GB]
I[Iceberg REST Catalog → Dell:8181]
S3[S3A → SeaweedFS Dell:8333]
end
J -->|build_spark_session| B
B -->|SPARK_CONNECT_URL set?| SC{ }
SC -->|yes| R[.remote sc://asus:15002]
SC -->|no| L[embedded .getOrCreate]
R -->|gRPC| S
S --> I
S --> S3
Implementation Details
In src/entsoe_pipeline/spark/spark_builder.py:
The existing build_spark_session() gains an environment-controlled branch. No existing configs, catalog settings, or JAR coordinates are removed — they remain active in the embedded path and are mirrored in the Spark Connect Server startup.
import os
from pyspark.sql import SparkSession
def build_spark_session(app_name: str = "ENTSOE_Lakehouse") -> SparkSession:
connect_url = os.environ.get("SPARK_CONNECT_URL")
if connect_url:
return (
SparkSession.builder
.remote(connect_url)
.appName(app_name)
.getOrCreate()
)
# Existing embedded Spark path — unchanged
hosts = get_hosts_config()
# ... (all current configs remain exactly as-is)
return SparkSession.builder.appName(app_name) \
.config("spark.jars.packages", ...) \
.getOrCreate()
On the Spark Master (ASUS):
The existing Spark cluster at /opt/spark/ (Master already running on :7077) is reused. The Spark Connect Server is started as a bridge to the existing Master, forwarding gRPC requests to the Workers on HP and Dell:
# On ASUS — start Spark Connect Server pointing to existing Master
/opt/spark/sbin/start-connect-server.sh \
--master spark://192.168.0.131:7077 \
--conf spark.connect.grpc.binding.port=15002 \
--packages "org.apache.iceberg:iceberg-spark-runtime-4.1_2.13:1.11.0,org.apache.iceberg:iceberg-aws-bundle:1.11.0,org.apache.hadoop:hadoop-aws:3.4.2" \
--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
--conf spark.sql.catalog.lakehouse=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.lakehouse.type=rest \
--conf spark.sql.catalog.lakehouse.uri=http://dell:8181/v1/${S3_TABLE_BUCKET}
The Iceberg/S3 catalog and S3A filesystem configs remain as-is in the existing cluster's spark-defaults.conf on ASUS — no changes needed there. The Connect Server inherits them from the Master.
Configuration
The switch is controlled by a single environment variable, consistent with the project's .env-for-secrets-only policy (ADR-001, ADR-002). The variable is set in the Kestra container's environment block in docker-compose.yml:
# In the kestra service definition:
environment:
- SPARK_CONNECT_URL=sc://asus:15002
For local development without Docker, the variable is left unset and build_spark_session() falls back to embedded Spark — no developer workflow change required.
The Spark Connect Server's JARs and catalog configurations are declared in docker-compose.yml under the spark-connect service command. They are an exact mirror of the spark.jars.packages and catalog configs in the embedded path of spark_builder.py. Both paths must be kept in sync when upgrading Iceberg/Spark versions — this is documented as a maintenance note in the Consequences section.
Testing
In tests/spark/test_spark_builder.py:
import os
import pytest
from unittest import mock
from entsoe_pipeline.spark.spark_builder import build_spark_session
class TestSparkSessionBuilder:
"""Verify Spark Connect vs embedded branching logic."""
def test_remote_session_when_connect_url_set(self, monkeypatch):
"""SPARK_CONNECT_URL → .remote() is called, no JVM configs."""
monkeypatch.setenv("SPARK_CONNECT_URL", "sc://localhost:15002")
with mock.patch("pyspark.sql.SparkSession.builder") as mock_builder:
mock_builder.remote.return_value = mock_builder
mock_builder.appName.return_value = mock_builder
build_spark_session("test_app")
mock_builder.remote.assert_called_once_with("sc://localhost:15002")
# Embedded configs must NOT be called
mock_builder.config.assert_not_called()
def test_embedded_session_when_no_connect_url(self):
"""No SPARK_CONNECT_URL → full embedded config path."""
# Ensure env var is absent
if "SPARK_CONNECT_URL" in os.environ:
del os.environ["SPARK_CONNECT_URL"]
# Cannot fully test .getOrCreate() without a JVM, but verify
# that the builder chain doesn't call .remote()
with mock.patch("pyspark.sql.SparkSession.builder") as mock_builder:
mock_builder.appName.return_value = mock_builder
mock_builder.config.return_value = mock_builder
try:
build_spark_session("test_app")
except Exception:
pass # Expected — no JVM in test environment
mock_builder.remote.assert_not_called()
mock_builder.config.assert_called() # Embedded configs must be invoked
def test_remote_session_preserves_app_name(self, monkeypatch):
"""App name propagates correctly in remote mode."""
monkeypatch.setenv("SPARK_CONNECT_URL", "sc://hp:15002")
with mock.patch("pyspark.sql.SparkSession.builder") as mock_builder:
mock_builder.remote.return_value = mock_builder
mock_builder.appName.return_value = mock_builder
build_spark_session("ENTSOE_Lakehouse")
mock_builder.appName.assert_called_once_with("ENTSOE_Lakehouse")
Consequences
Positive Outcomes
- Reduces Kestra orchestrator image size from ~4.9 GB to ~300 MB, cutting build and deploy time by approximately 16×.
- Aligns with existing multi-machine topology: Spark Master on ASUS, Workers on HP (6C/10GB) and Dell (4C/4GB) — all already running in ALIVE state. Zero new containers needed.
- Decouples Spark JVM lifecycle from Kestra — Spark cluster and Connect Server run independently of the orchestrator.
- Zero changes required in business logic — all Spark-consuming functions already accept
SparkSessionas a parameter. - Embedded fallback preserved for local development with zero configuration.
Negative Consequences / Trade-offs
- Introduces a Spark Connect Server process on ASUS to manage alongside the existing Master process.
- Spark Connect Server JAR/config must be kept in sync with the existing cluster's
spark-defaults.confon ASUS and the embedded path inspark_builder.py. - gRPC adds ~2-5 ms network overhead per DataFrame operation (acceptable on same-LAN deployment).
- Spark Connect does not support all legacy RDD APIs — not relevant here as the project uses only DataFrame API, which is fully supported.
- Debugging Spark failures now requires checking two log sources: Kestra flow logs (client side) and Spark Connect Server logs (server side).
Ongoing Maintenance & Considerations
- When upgrading
iceberg-spark-runtimeorhadoop-awsversions, update: (1) ASUSspark-defaults.conf(cluster), (2)spark_builder.py(embedded fallback). The Connect Server inherits configs from the Master. - Monitor gRPC on ASUS:15002 — Spark Connect client raises
grpc.RpcErroron unavailability, caught by the existing DLQ pattern. - The Spark Connect Server on ASUS should be added to the Grafana dashboard alongside the existing Master (already visible on
:8080). - Consider adding a health-check endpoint (
/health→spark.range(1).count()) to the Spark Connect Server for Uptime Kuma monitoring.
Dependencies
- Infrastructure: Existing Spark standalone cluster (ASUS Master + HP/Dell Workers, 10 cores / 14 GB), Docker LAN for Kestra-to-ASUS gRPC.
- Data Frameworks:
pysparkwith connect extra >= 4.0.0,apache-iceberg >= 1.11.0(via Spark JARs on server),grpcio(transitive dependency). - Removed from orchestrator image:
pysparkwith sql extra JVM,org.apache.iceberg:iceberg-spark-runtime,org.apache.hadoop:hadoop-aws(moved to Spark Connect Server).
References
- Spark Connect Overview (Apache Spark 4.1) — Client/server architecture and gRPC protocol specification.
- Apache Iceberg Spark Integration — Catalog and FileIO configuration for Spark sessions.
- ADR-001: Centralized YAML Configuration — Config SSOT pattern;
SPARK_CONNECT_URLfollows the.env-for-secrets-only rule. - ADR-004: Ephemeral Landing Event-Driven Staging — Staging ingestion flow that invokes Spark jobs.
- ADR-007: Separation of Metadata Refresh from Ingestion — Fault isolation and DLQ pattern reused for Spark Connect failures.
Changelog
- v1.0 (2026-07-23): Initial proposed version. Spark Connect split design with embedded fallback.
Title
Migration from Kestra (YAML-based) to Apache Airflow (Python-based) for Pipeline Orchestration
Description
We will replace the Kestra orchestrator with Apache Airflow, converting all three Kestra YAML flows into Python DAGs while maintaining identical job execution logic. Airflow will run in a Docker container alongside the project codebase mounted as volumes, eliminating the Docker-in-Docker pattern required by Kestra.
Context
The ENTSO-E pipeline currently uses Kestra v1.3 (Java-based, YAML-defined flows) running in a Docker container. Each Kestra flow executes Python jobs inside a separate entsoe-pipeline:latest Docker image via Kestra's Docker task runner. This architecture creates several practical problems:
-
Docker-in-Docker overhead: Kestra (Java container) spawns Python jobs inside a second container (
entsoe-pipeline:latest). Two Docker daemon layers, two image registries, two networking stacks. -
Full image rebuild on every code change: Because job code is baked into the
entsoe-pipeline:latestimage (viaCOPY), any change tojobs/,src/, orconfig/requires a full Docker rebuild. Withuv syncand layer caching this takes 30–60 seconds — unacceptable for rapid iteration. -
YAML verbosity: The
refresh_fms_metadataflow requires 180+ lines of YAML to express what would be a 15-line Python loop over domains and environments. Conditional logic ({% if inputs.test %}) is brittle Jinja2 templating embedded in YAML strings. -
Image size: The Kestra image (
kestra/kestra:v1.3) is 6.89 GB, mostly JVM and bundled plugins. This increases pull times, disk usage, and cold-start latency. -
Limited ecosystem: Kestra's plugin marketplace and community documentation are significantly smaller than Airflow's. Finding answers to operational questions (SLA configuration, alerting, metric export) requires reading Java source code.
In contrast, Apache Airflow is the de-facto industry standard for Python-based data pipeline orchestration. It has 45,000+ GitHub stars, 17,000+ forks, a massive plugin ecosystem, and native Python DAG authoring.
Decision Drivers
- Developer velocity: Code changes must be reflected in <30 seconds without image rebuilds.
- Image size: The orchestrator image should consolidate both the scheduler AND job execution into a single image, eliminating the separate
entsoe-pipelineimage required by Kestra. - Authoring ergonomics: DAGs must be Python code with IDE autocompletion, type checking, and standard control flow (loops, functions).
- Industry standard: Using Airflow signals competence to potential employers (TradingView, European contractors).
- Observability: Native integration with our existing Grafana + Loki stack.
Alternatives
-
Option A (Keep Kestra): Continue with Kestra, accepting the Docker rebuild penalty and 6.89 GB image.
Pros: Zero migration effort. Cons: All problems listed above remain. Image rebuild cycle slows iteration. -
Option B (Celery + Redis): Use Celery as a Python-native task queue with Redis broker.
Pros: Lighter than Airflow, pure Python. Cons: No built-in DAG scheduling UI, no SLA tracking, no retry history dashboard. Would need to build a frontend from scratch. -
Option C (Apache Airflow in Docker): Replace Kestra with Airflow, running in a Docker container with project code mounted as volumes.
Pros: Industry standard, Python DAGs, no Docker-in-Docker, <30s code refresh, single image replaces Kestra + entsoe-pipeline. Cons: One-time migration of 3 Kestra flows to Python DAGs (~2 hours).
Decision Framework
| Criterion (Weight) | Kestra | Celery+Redis | Airflow |
|---|---|---|---|
| Developer Velocity (30%) | 2 | 4 | 5 |
| Industry Standard (25%) | 2 | 2 | 5 |
| Image Consolidation (15%) | 1 | 5 | 4 |
| Observability (15%) | 3 | 2 | 5 |
| Migration Effort (15%) | 5 | 3 | 4 |
| Total | 2.55 | 3.20 | 4.75 ✅ |
Decision
We will adopt Apache Airflow running in a Docker container with the apache/airflow:latest base image. The project source code (dags/, jobs/, src/, config/, config_env/) will be mounted as Docker volumes, not baked into the image. Only Python dependencies (pyproject.toml → uv sync --frozen --no-dev --no-install-project) are installed at build time.
The three existing Kestra flows will be converted to Python DAGs:
- dags/landing.yml → dags/landing.py
- dags/refresh_fms_metadata.yml → dags/refresh_fms_metadata.py
- dags/staging.yml → dags/staging.py
Kestra containers (kestra, kestra_postgres) are marked with the kestra Compose profile in docker/docker-compose.yml and will not start by default. They will be fully removed after the Airflow deployment is validated.
High-Level Architecture
graph TD
A[Apache Airflow Scheduler] --> B[Airflow Webserver :8084]
A --> C[Airflow Postgres]
A --> D[DAGs directory /opt/airflow/dags]
D --> E[PythonOperator: jobs/landing/ingest_my_entsoe_domains.py]
D --> F[PythonOperator: jobs/staging/ingest_landing_csv_to_lakehouse.py]
D --> G[PythonOperator: jobs/refresh_fms_metadata.py]
subgraph "Docker Volume Mounts (host)"
H[dags/]
I[jobs/]
J[src/]
K[config/]
L[config_env/]
M[.data/logs/]
end
H -.-> D
I -.-> E
J -.-> E
K -.-> E
L -.-> E
M -.-> N[Promtail → Loki → Grafana]
Related Requirements
Functional Requirements
- FR-1: The system must execute all three existing pipeline jobs (landing preparation, domain ingestion, staging to Iceberg) with identical behavior as Kestra.
- FR-2: The Airflow webserver UI must be accessible for DAG monitoring and manual triggering.
Non-Functional Requirements
- NFR-1 (Image consolidation): A single Airflow image must replace both the Kestra orchestrator (6.89 GB) AND the separate
entsoe-pipelinejob-execution image. - NFR-2 (Code refresh): Changes to DAGs, jobs, or source code must be reflected without image rebuild.
- NFR-3 (Path consistency): All host paths in Docker volumes must use the SSOT path system (
paths.yml→paths.py→justfileexport → Compose${VAR}).
Integration Requirements
- IR-1: The
RunsLoggermust detect Airflow'sAIRFLOW_CTX_DAG_RUN_IDenvironment variable (falling back toKESTRA_EXECUTION_IDfor transition period). - IR-2: Pipeline logs must continue flowing through
setup_logging(use_json=True)→.data/logs/→ Promtail → Loki → Grafana.
Performance Requirements
- PR-1: Airflow DAG parsing interval must be ≤30 seconds to ensure code changes are reflected promptly during development.
- PR-2: The Airflow scheduler must handle all three DAGs concurrently without task queue backpressure under normal operating conditions.
Related Decisions
- ADR-001 (Centralized YAML Configuration): Airflow port and path configurations are stored in
config_env/ports.ymlandconfig/paths.yml. - ADR-005 (Pull-Request-Driven Git Workflow): DAG changes follow the same feature-branch workflow as all other code.
Design
Architecture Overview
The Airflow deployment uses a single custom Dockerfile based on apache/airflow:latest:
FROM apache/airflow:latest
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project
ENV PYTHONPATH=/opt/airflow/src:/opt/airflow:$PYTHONPATH
Three services in docker/airflow/docker-compose.yml: airflow-postgres (metadata DB), airflow-scheduler (DAG execution), airflow-webserver (UI on port 8084).
All project directories are mounted as volumes using SSOT-exported environment variables (${DAGS_DIR}, ${JOBS_DIR}, ${SRC_DIR}, etc.). No hardcoded paths.
Implementation Details
In dags/landing.py:
from airflow import DAG
from airflow.operators.python import PythonOperator
def _prepare_ingestion() -> None:
from jobs.landing.prepare_landing_ingestion import main
main()
def _ingest_domains() -> None:
from jobs.landing.ingest_my_entsoe_domains import main
main()
with DAG(dag_id="landing", schedule="@daily", ...):
prepare_ingestion = PythonOperator(task_id="prepare_ingestion", python_callable=_prepare_ingestion)
ingest_domains = PythonOperator(task_id="ingest_domains", python_callable=_ingest_domains)
prepare_ingestion >> ingest_domains
The refresh_fms_metadata DAG uses a Python loop over DOMAINS × ENVIRONMENTS (18 tasks) instead of Kestra's 200+ lines of YAML. Tasks are flat (not nested in TaskGroups) for maximum scheduler parallelism.
In src/entsoe_pipeline/logger/core/runs_tracker.py:
self.run_id = (
os.environ.get("AIRFLOW_CTX_DAG_RUN_ID")
or os.environ.get("KESTRA_EXECUTION_ID") # backward compat during transition
or str(uuid.uuid4())
)
Configuration
In config_env/ports.yml:
airflow_webserver: 8084
In config/paths.yml:
DAGS_DIR: "dags"
JOBS_DIR: "jobs"
In justfile:
airflow_webserver_port := `uv run python -c "from entsoe_pipeline import get_ports_config; print(get_ports_config().airflow_webserver)"`
export AIRFLOW_WEBSERVER_PORT := airflow_webserver_port
airflow-up:
docker compose --env-file .env -f docker/airflow/docker-compose.yml up -d --build
Testing
Integration test for Airflow DAG integrity:
import pytest
from airflow.models import DagBag
def test_all_dags_load_without_errors() -> None:
"""Verify every DAG in dags/ parses without import or syntax errors."""
dag_bag = DagBag(dag_folder="dags/", include_examples=False)
assert len(dag_bag.import_errors) == 0, (
f"DAG import errors: {dag_bag.import_errors}"
)
assert dag_bag.dags, "No DAGs were loaded"
@pytest.mark.parametrize("dag_id", ["landing", "staging", "refresh_fms_metadata"])
def test_dag_task_count(dag_id: str) -> None:
"""Verify each DAG has the expected number of tasks."""
dag_bag = DagBag(dag_folder="dags/", include_examples=False)
dag = dag_bag.get_dag(dag_id)
assert dag is not None, f"DAG '{dag_id}' not found"
task_count = len(dag.tasks)
expected = {"landing": 2, "staging": 1, "refresh_fms_metadata": 20}
assert task_count == expected[dag_id], (
f"DAG '{dag_id}' has {task_count} tasks, expected {expected[dag_id]}"
)
Consequences
Positive Outcomes
- Eliminates Docker rebuild cycle: Code changes appear in Airflow within 30 seconds (DAG parsing interval). No image rebuild needed.
- Consolidates orchestrator and executor: A single 7.79 GB Airflow image (base
apache/airflow:3.3.0+ 2.55 GB Python deps fromuv sync) replaces the Kestra orchestrator (6.89 GB) AND the separateentsoe-pipelinebuild image. Theuv synclayer is cached and only rebuilds whenuv.lockchanges. - Actual image size: 7.79 GB. The
uv synclayer (2.55 GB) is dominated by PySpark Connect JAR dependencies. However, this single image eliminates the Docker-in-Docker pattern completely — no second container, no second image build cycle. - Python-native DAG authoring: IDE autocompletion, mypy type checking, standard control flow, git-diff friendly.
- Industry-standard skill: Airflow experience is listed on the TradingView job description and is a core competency for data engineering roles.
- Seamless Grafana integration: Airflow metrics can be exported to Prometheus and visualized alongside pipeline logs in our existing Grafana instance.
Negative Consequences / Trade-offs
- One-time migration effort: Three Kestra YAML flows must be converted to Python DAGs (~2 hours of work).
- Learning curve: Airflow concepts (DAGs, Operators, XComs, Connections) differ from Kestra's task-runner model.
- Additional Postgres instance: Airflow requires its own metadata database (separate from
entsoe_postgres). This adds one lightweight Postgres container (~50 MB RAM).
Ongoing Maintenance & Considerations
- Monitor Airflow scheduler health via
airflow jobs checkin a healthcheck script. - Rotate Airflow Postgres password periodically.
- Airflow DAG parsing interval (default 30s) should remain as-is for development; reduce to 5s in production if faster DAG refresh is needed.
- Old Kestra volumes (
docker_kestra_data,docker_kestra_postgres_data) can be removed after successful Airflow validation. - Kestra image (6.89 GB) should be removed to free disk space.
Dependencies
- Infrastructure: Docker, Docker Compose.
- Orchestration:
apache/airflow >= 2.10. - Python:
apache-airflowpackage (installed in container, not in projectpyproject.toml). - Removed:
kestra/kestra:v1.3image,kestra_postgrescontainer,docker/kestra.yml.
References
- Apache Airflow Documentation — Core concepts, DAG authoring, operators.
- Airflow Docker Image — Official Docker/Podman images.
- GitHub MCP Server — Used for AI-assisted PR creation for DAG changes.
- Kestra Documentation — Reference for the system being migrated away from.
- ADR-001: Centralized YAML Configuration — SSOT config pattern used for Airflow ports.
- ADR-005: Pull-Request-Driven Git Workflow — DAG changes follow this workflow.
Changelog
- v1.0 (2026-07-30): Initial accepted version. Migration completed: 3 DAGs converted, Kestra deprecated.
Postmortems (2)
Summary
DAG refresh_fms_metadata exhausted 16 GB RAM + 10 GB swap within 2 minutes of startup, sending the HP host into swap thrashing. Two independent root causes were found, both triggered by 18 parallel Airflow tasks (2 environments × 9 domains):
-
Root Cause A (first fix):
all_env_file_details— an unbounded list accumulating all file-metadata dicts across all folders in a domain. Used only for a one-line log summary. -
Root Cause B (second fix, discovered after A was patched but OOM persisted): 18 parallel tasks each independently downloaded
Export_log_r3.csv+Export_oce_log_r3.csvinto pandas DataFrames — 18 copies of the same multi-hundred-MB CSVs in RAM. Plus each task calledfetchall()on the fullfms_filestable.
The first fix reduced memory from path A, but path B remained untouched — the host still swap-thrashed. The DAG was redesigned to download logs once, then distribute only lightweight crawl tasks.
Final result (verified 2026-08-01 14:30 MSK):
| Metric | Before | After |
|---|---|---|
| Peak RAM | 16 GB → 100%, swap 10 GB → 100% | 8 GB peak, settled 6.5 GB (base 4 GB + 2.5 GB) |
| Swap | 10 GB exhausted | < 1 MB |
| OOM kills | Multiple workers killed per run | Zero |
| Delta analysis (IOP) | N/A (was embedded in crawl) | 12 seconds |
| Delta analysis (PROD) | N/A | 1 min 15 sec |
| Crawl (heaviest: PROD Market) | ~10+ min with swap thrashing | 2 min 10 sec |
| Idempotency | Broken by OOM kills | Fully idempotent |
Timeline
| Time (UTC+3) | Event |
|---|---|
| ~11:20 | DAG refresh_fms_metadata started |
| ~11:21 | RAM → 100%, swap begins filling |
| ~11:22 | Swap exhausted (10 GB). OOM-killer kills Airflow workers |
| ~11:23 | Cycle: OOM → kill → restart → OOM |
| ~11:25 | Stan notices degradation, notifies Werner |
| ~11:30 | Root Cause A found: all_env_file_details unbounded list |
| ~11:45 | Fix A deployed: running totals instead of a list (8 files changed) |
| ~11:50 | DAG restarted — OOM STILL HAPPENS. Fix A was insufficient |
| ~14:00 | Werner reads the full delta refresh code path |
| ~14:15 | Root Cause B found: 18× pandas DataFrames from identical log CSVs |
| ~14:45 | Fix B designed: extract delta analysis to a single phase, split DAG into 4 phases |
| ~15:00 | Fix B deployed. DAG restarted — RAM peaked at 8 GB, swap < 1 MB. Pipeline healthy. |
| ~15:10 | Second run (idempotency check): delta analysis in seconds, crawls skip (all fresh). Confirmed idempotent. |
Detection
- Primary signal: Grafana
Cluster Nodesdashboard — RAM usage instantly reached 100%, swap filled completely. - Secondary signals: Airflow Web UI showed tasks cycling
queued→running→killed→up_for_retry;dmesgcontained OOM killer entries. - Alerting: No swap alert existed. This is a monitoring gap.
Root Cause
Root Cause A: Unbounded list in full-scan path
all_env_file_details in ingest_domain_metadata() (domain_metadata_crawler.py) accumulated ALL file dicts from ALL folders in a domain:
all_env_file_details = [] # ← grows without bound
for folder in folders:
files = crawl_metadata_folder(...)
all_env_file_details.extend(files) # ← every dict stays in memory
Used only once — in finalize_crawl_session() → compile_env_stats() — for a log line. 18 parallel processes × ~150,000 dicts × ~500 bytes = ~1.35 GB just for these lists.
Fix: replaced with 5 running-total variables (total_files, total_original_bytes, total_compressed_bytes, min_month, max_month). Files changed:
| File | Change |
|---|---|
src/.../domain_metadata_crawler.py |
all_env_file_details: list → 5 running-total variables |
src/.../legacy_metadata_crawler.py |
Same |
src/.../metadata_crawlers_common.py |
finalize_crawl_session signature updated |
src/.../transformer.py |
compile_env_stats now receives aggregates |
Root Cause B: 18× pandas DataFrames in delta-refresh path
After Fix A, the OOM persisted. The second root cause was in run_delta_metadata_refresh() (fms_metadata_delta.py). Each of the 18 parallel tasks independently:
-
Downloaded two multi-hundred-MB CSVs into pandas DataFrames:
python df_std = client.download_single_file("", "Export_log_r3.csv") # ~100-200 MB → pandas df_oce = client.download_single_file("", "Export_oce_log_r3.csv") # ~100-200 MB → pandas -
Called
fetchall()on the fullfms_filestable:python db_file_rows = conn.execute(select(...)).fetchall() # millions of rows -
Built intermediate lists (
std_records,oce_records, dicts fordb_files,folder_crawled_at).
The two CSV files are identical for all 18 tasks within the same environment, yet each task downloaded and parsed them independently. Memory breakdown:
| Per-task item | Size | × 18 |
|---|---|---|
| Two pandas DataFrames (CSV + overhead) | ~400 MB | 7.2 GB |
fetchall() — all fms_files rows |
~100 MB | 1.8 GB |
| Python process + SQLAlchemy engine | ~100 MB | 1.8 GB |
| Total | ~600 MB | ~10.8 GB |
Fix: redesigned the DAG from 3 phases to 4 phases, inserting a new analyze_delta phase that runs ONCE per environment:
Before: prepare → 18× crawl(env, domain) → finalize
After: prepare → 2× analyze_delta(env) → 18× crawl_dirty_only(env, domain) → finalize
New files:
| File | Purpose |
|---|---|
jobs/fms_metadata/analyze_fms_delta.py |
Downloads logs ONCE per env, writes dirty-folder list to .data/dirty_folders/<env>.json |
jobs/fms_metadata/crawl_dirty_folder.py |
Reads JSON, crawls only assigned dirty leafs. No pandas, no fetchall() |
dags/refresh_fms_metadata.py |
Updated to 4-phase DAG |
config/paths.yml |
Added DIRTY_FOLDERS_DIR: ".data/dirty_folders" |
Memory after fix:
| Phase | Memory |
|---|---|
analyze_delta (2 tasks, sequential) |
~400 MB each |
crawl_dirty_only (18 tasks, parallel) |
~80 MB each = 1.4 GB total |
| Peak | ~2 GB (vs ~10 GB before) |
Why was it not caught earlier?
- Swap monitoring was missing in Grafana — no
node_memory_SwapFree_bytespanel - The first fix (A) looked correct and was code-reviewed; nobody re-tested with a full parallel run before declaring victory
- The delta-refresh path downloads logs before checking whether a full scan fallback is needed — the pandas DataFrames are always loaded even for lightweight incremental runs
- The code path that exploded (log download) was in a different file (
fms_metadata_delta.py) from the code that was fixed (domain_metadata_crawler.py)
Impact
| Metric | Value |
|---|---|
| Duration | ~30 minutes of degraded DAG execution (first run); persisted across multiple runs until Fix B |
| Data loss | None — pipeline is idempotent, data intact |
| Memory peak | 16 GB RAM → 100%, 10 GB swap → 100% |
| Affected DAG runs | 2+ runs of refresh_fms_metadata in swap-thrashing state |
| Processes killed | Multiple Airflow workers OOM-killed, automatic Airflow restart |
| User-facing impact | None — internal metadata pipeline |
Resolution
Immediate actions
During diagnostics, the DAG was not stopped — tasks were killed and restarted by Airflow, and the pipeline slowly crawled to completion through swap.
Permanent fix
Fix A (11:45): all_env_file_details → running totals. Reduced memory in the full-scan path but did NOT fix the delta-refresh path.
Fix B (14:45): DAG redesigned — delta analysis extracted to a single non-parallel phase. The 18 parallel tasks no longer download logs or call fetchall().
Prevention
- [ ] Add swap usage panel to Grafana
Cluster Nodesdashboard - [ ] Add Prometheus alert:
node_memory_SwapFree_bytes < 1GB→ Mattermost/entsoe-alerts - [ ] Add pre-push test: run
refresh_fms_metadatawith 2 parallel tasks and verify memory stays under 4 GB - [ ] Code review rule: flag any
fetchall()without LIMIT or pagination - [ ] Code review rule: flag any large file download inside a parallel task that could be shared
Prevention
Monitoring improvements
- [ ] Add swap usage panel to Grafana
Cluster Nodesdashboard - [ ] Add Prometheus alert:
node_memory_SwapFree_bytes < 1GB→ Mattermost/entsoe-alerts - [ ] Add per-worker memory metrics to Airflow monitoring
Code / process improvements
- [ ] Add pre-push integration test: run DAG with 2 parallel domains, assert RAM < 4 GB
- [ ] Code review checklist: flag
fetchall()without LIMIT, flag.extend()in unbounded loops - [ ] Code review checklist: flag any file download inside a parallel task that could be shared across workers
Documentation
- [ ] Postmortem filed:
docs/postmortem/PM-001-oom-refresh-fms-metadata.md - [ ] Document the two-root-cause pattern in
.agents/rules/— when a fix doesn't resolve the incident, re-audit the ENTIRE execution path, not just the patched function
Lessons Learned
-
A single incident can have multiple independent root causes. After fixing the obvious memory leak in the full-scan path, the OOM persisted — forcing a second, deeper audit of the delta-refresh path. Always verify the fix under production load before closing.
-
Parallel tasks downloading identical data is a silent memory bomb. 18 workers × the same CSV = 18 copies in RAM. The design smell: if two tasks in a DAG need the same input, that input should be a separate upstream task.
-
Swap is the canary for memory leaks. RAM spikes are hard to catch in real time; swap exhaustion is visible and slow. A Prometheus alert on
node_memory_SwapFree_byteswould have caught this minutes before the OOM killer activated. -
fetchall() on a growing table will eventually OOM regardless of parallelism. Even a single worker loading millions of
fms_filesrows into a Python list is unsustainable. Future-proof with server-side cursors or paginated queries.
References
- Grafana dashboard:
grafana.stan-buren.ru→ENTSOE→Cluster Nodes - DAG:
dags/refresh_fms_metadata.py - Delta refresh:
src/entsoe_pipeline/fms_metadata/core/fms_metadata_delta.py→run_delta_metadata_refresh() - Full scan:
src/entsoe_pipeline/fms_metadata/core/domain_metadata_crawler.py→ingest_domain_metadata() - New delta analysis:
jobs/fms_metadata/analyze_fms_delta.py - New lightweight crawler:
jobs/fms_metadata/crawl_dirty_folder.py - Paths SSOT:
config/paths.yml→DIRTY_FOLDERS_DIR
Summary
After the July 31 network migration and August 1 host reboot, the staging DAG — which reads CSV files from the SeaweedFS S3 landing zone and merges them into Apache Iceberg staging tables via Spark — failed silently on every execution. Tasks hung for 60+ seconds at "Reading CSV path(s)" then were killed by Airflow (exit 143). The root cause was NOT a single bug, but three independent failures across the Spark stack: (1) AWS SDK JAR version conflicts in the classpath, (2) the VPN interface capturing the Spark driver's bind address, and (3) default executor memory capped at 1 GB per worker. All three had to be fixed simultaneously; fixing any single layer left the pipeline broken.
Timeline
| Time (UTC+3) | Event |
|---|---|
| ~13:30 | Staging DAG first run after reboot — hangs at "Reading 3 CSV path(s)" |
| ~13:33 | Airflow kills task after 60 seconds (exit 143) |
| ~13:32–15:03 | Multiple attempts — same ObjectTransfer error. JARs manually copied, packages reordered, Ivy cache cleared — no effect |
| ~15:09 | Read official Spark + Iceberg + SeaweedFS documentation. Identified correct package combination (matching versions, no hadoop-aws in --packages) |
| ~15:15 | Deployed correct packages. Error persists — JAR fix alone insufficient |
| ~16:19 | Stan inspects Dell executor stderr — discovers 10.0.85.1:42227 connection timeout. VPN IP is capturing the Spark driver address |
| ~16:35 | Added SPARK_LOCAL_IP + spark.driver.host + spark.driver.bindAddress to all three machines. Spark master/connect/workers restarted |
| ~16:37 | Added s3-transfer-manager JAR manually to ASUS spark/jars for the missing ObjectTransfer class |
| ~16:39 | Configured executor memory: spark.executor.memory=2g, spark.executor.cores=2, worker limits via host_vars (HP: 6c/7g, Dell: 4c/5g) |
| ~18:37 | Pipeline succeeded. 3 CSV files (38 MB) → Iceberg MERGE in 87 seconds. Metrics: processed=3, errors=0 |
Detection
- Primary signal: Airflow task logs —
SparkConnectGrpcException: ObjectTransferafter every run - Secondary signal: Dell worker stderr —
Connecting to /10.0.85.1:42227 timed out (120000 ms)— VPN IP unreachable from worker - Tertiary signal: Spark Master UI — 22 finished executors (KILLED/EXITED) accumulated across today's failed runs
- Alerting: No alerting was configured for Spark executor failures or DAG task timeouts. This is a monitoring gap.
Root Cause
Root Cause A: AWS SDK JAR version conflict
The spark-connect.service was deployed with three packages pulling incompatible versions of the AWS SDK:
hadoop-aws:3.4.2 → AWS SDK V2, SHADED (classes relocated)
iceberg-aws-bundle:1.11.0 → AWS SDK V2, unshaded (different version)
s3-transfer-manager:2.29.52 → AWS SDK V2, unshaded (third version)
hadoop-aws:3.4.2 (Hadoop 3.4 upgraded S3A to AWS SDK V2) uses a shaded bundle where all software.amazon.awssdk.* classes are relocated. iceberg-aws-bundle:1.11.0 bundles an unshaded AWS SDK that predates the S3 Transfer Manager module. The s3-transfer-manager:2.29.52 JAR provides the ObjectTransfer class but its base AWS SDK classes conflict with the older bundled version. Result: ClassNotFoundException: software.amazon.awssdk.transfer.s3.model.ObjectTransfer — the class was either absent (from the old bundle) or its dependencies were incompatible (shaded vs unshaded).
Fix: Removed hadoop-aws and s3-transfer-manager from --packages. Kept only iceberg-spark-runtime:1.11.0 + iceberg-aws-bundle:1.11.0 (matching versions, per official Iceberg + SeaweedFS docs). Copied hadoop-aws-3.4.2.jar and s3-transfer-manager-2.29.52.jar manually to /opt/spark/jars/ on all machines — placed directly on the classpath without Maven transitive dependency resolution.
Root Cause B: VPN binding the Spark driver address
ASUS hostname (asus-25-498) resolves to loopback (127.0.1.1). Spark falls back to a non-loopback interface and picks 10.0.85.1 — the outline-tun0 VPN interface. The Spark driver binds to 10.0.85.1:42227. Workers on HP (192.168.0.93) and Dell (192.168.0.210) are on the local network and cannot reach 10.0.85.1. Result: executor connection timeout after 120 seconds, executors killed, tasks never execute. 22 executors were created and killed across today's failed runs.
Fix: Added SPARK_LOCAL_IP environment variable to all Spark systemd service files (master, connect, workers). Added spark.driver.host and spark.driver.bindAddress to the Spark Connect server configuration. All bound to 192.168.0.x (local network), bypassing the VPN interface entirely.
Root Cause C: Default executor memory capped at 1 GB
Spark's default spark.executor.memory=1g created one executor per worker with all available cores but only 1 GB memory. The cluster had 14 GB RAM available (HP: 10 GB, Dell: 4 GB) but only 2 GB was used. This didn't block execution but slowed it significantly and risked OOM on larger datasets.
Fix: Set spark.executor.memory=2g and spark.executor.cores=2. Limited worker capacity via spark.worker.cores and spark.worker.memory per host (HP: 6c/7g → 3 executors, Dell: 4c/5g → 2 executors). Total: 10 GB across 5 executors.
Why was it not caught earlier?
- The VPN network migration (July 31) moved the VPN from ASUS to HP, but ASUS's VPN interface (
outline-tun0) remained active and Spark was not configured to ignore it - JAR version conflicts were silent —
--packagesresolved and downloaded all artifacts successfully; the classloader failure only manifested at runtime when Spark Connect'sanalyzePlantried to instantiate S3 Transfer Manager classes - No integration test existed for the staging DAG after infrastructure changes
- Spark executor memory was never explicitly configured — the default 1 GB worked for tiny test datasets and was never challenged with production data
Impact
| Metric | Value |
|---|---|
| Duration | ~5 hours of failed staging DAG execution |
| Data loss | None — pipeline is idempotent; CSV files remained in landing zone; ingestion attempt logs preserved in PostgreSQL |
| Failed DAG runs | ~20 runs of staging (mix of scheduled and manual triggers) |
| Executors killed | 22 executors across multiple Spark Connect application instances |
| Root causes | 3 independent failures (JARs, network, memory) |
| User-facing impact | None — internal pipeline; landing DAG continued to download CSVs successfully |
Resolution
Immediate actions
- Manually copied
hadoop-awsands3-transfer-managerJARs to/opt/spark/jars/on ASUS - Added
SPARK_LOCAL_IPto all systemd service files - Configured executor memory and worker limits
Permanent fix
Files changed:
| File | Change |
|---|---|
infra/ansible/group_vars/all.yml |
Reduced packages to 2 (matching versions 1.11.0); added worker defaults |
infra/ansible/host_vars/hp.yml |
Worker limits: 6 cores / 7 GB |
infra/ansible/host_vars/dell.yml |
Worker limits: 4 cores / 5 GB |
infra/ansible/templates/spark-defaults.conf.j2 |
Added spark.worker.*, spark.executor.*, SPARK_LOCAL_IP |
infra/ansible/templates/spark-connect.service.j2 |
Added SPARK_LOCAL_IP, spark.driver.host, spark.driver.bindAddress |
infra/ansible/inventory.local.yml |
Added ansible_connection: local for HP |
Deploy command: just cluster-deploy — deploys spark-defaults.conf to all machines, spark-connect.service to ASUS, idempotent.
Prevention
Monitoring improvements
- [ ] Add Spark executor failure alert to Grafana
Cluster Nodesdashboard - [ ] Add DAG task timeout alert (> 60 seconds in "Reading CSV" phase → notification)
- [ ] Monitor
SPARK_LOCAL_IPbinding — alert if driver binds to non-192.168.x.x address
Code / process improvements
- [ ] Add pre-push integration test: run staging DAG with a single test CSV, assert Spark DataFrame loads in < 30 seconds
- [ ] After any network reconfiguration, run
just cluster-deployto ensure Spark configs are consistent - [ ] Document Spark JAR dependency policy: ONLY specify iceberg packages in
--packages; all Hadoop/S3 JARs go in/opt/spark/jars/manually to avoid Maven dependency conflicts
Documentation
- [ ] Postmortem filed:
docs/postmortem/PM-002-staging-spark-cluster-failure.md - [ ] Updated
.agents/rules/apache-airflow-knowledge.md— Spark diagnostics checklist - [ ] Updated
infra/README.md— Spark worker diagnostics section - [ ] Updated
.agents/rules/how-to-configure-infra-via-ansible.md— Spark cluster diagnostics, correct package versions
Lessons Learned
-
A single pipeline failure can have multiple independent root causes across the stack. The staging DAG failed because JARs, network, and memory were ALL broken simultaneously. Fixing any one layer left the others blocking. The diagnostic pattern: work from the top of the stack down (code → Spark Connect → cluster → network → OS), not sideways.
-
VPN interfaces MUST be explicitly excluded from Spark's interface detection.
SPARK_LOCAL_IPis not optional when the host has tunnel interfaces. If Spark can see a VPN IP, it WILL bind to it — and workers on the local network won't be able to reach it. -
Maven
--packagesis dangerous when multiple artifacts pull the same dependency in different versions. The classloader has no way to reconcile shaded Hudson AWS SDK with unshaded Iceberg AWS SDK. The only safe approach: use--packagesONLY for Iceberg (matching versions), and place all Hadoop/S3 JARs directly on the classpath. -
Spark defaults are for laptops, not clusters. 1 GB executor memory is the default for a reason — it works on a single machine. On a 3-node cluster with 14 GB RAM, it leaves 86% of resources unused. Always configure
spark.executor.memoryandspark.executor.coresexplicitly.
References
- SeaweedFS Spark Iceberg Integration: https://github.com/seaweedfs/seaweedfs/wiki/Spark-Iceberg-Integration
- Iceberg Spark Getting Started: https://iceberg.apache.org/docs/latest/spark-getting-started/
- Hadoop S3A SDK V2 Upgrade: https://hadoop.apache.org/docs/current/hadoop-aws/tools/hadoop-aws/aws_sdk_upgrade.html
- Spark Master UI: http://192.168.0.131:8080
- Grafana dashboard:
grafana.stan-buren.ru→ENTSOE→Cluster Nodes - Ansible spark playbook:
infra/ansible/spark.yml - Spark session builder:
src/entsoe_pipeline/spark/spark_builder.py
Root Cause Analyses (1)
Problem Statement
At 03:00 MSK on 2026-08-05, a MERGE INTO operation on Iceberg table energyprices_12_1_d_r3 failed with MERGE_CARDINALITY_VIOLATION, causing the Spark Connect session app-20260802111013-0002 to deadlock. The application has been stuck in RUNNING state for 72+ hours, consuming 10 cores and 10 GiB RAM without making progress.
Event Chain
2026-08-02 11:10 → Application app-20260802111013-0002 launched (Spark Connect server, user donald_trump)
2026-08-02 11:10 → 10 cores allocated across 2 workers: HP (6 cores, 6 GiB) + Dell (4 cores, 4 GiB)
...
2026-08-05 03:00 → Stage 44.0 begins: MERGE INTO energyprices_12_1_d_r3 via Iceberg SparkWrite
2026-08-05 03:00 → Executor 3 (Dell) encounters MERGE_CARDINALITY_VIOLATION on tasks 249 and 250
2026-08-05 03:00 → Driver cancels Stage 44.0 — reason: "ON condition matched single target row with multiple source rows"
2026-08-05 03:00 → Iceberg DataAndMetadataWritingSparkTask enters abort path (finally block)
2026-08-05 03:00 → abort commit calls S3FileIO.deleteFile() for staging cleanup
2026-08-05 03:00 → AWS SDK InterruptMonitor detects Thread.interrupt() flag → SdkInterruptedException
2026-08-05 03:00 → Iceberg retry logic (Tasks.java:413) re-attempts delete — immediately interrupted again
2026-08-05 03:00 → Executor 3 logs ERROR: "Aborted commit for partition 7 (task 250, attempt 2)" and goes silent
2026-08-05 03:01 → SerializableTableWithSize: "Releasing resources" × 3 — no further activity
2026-08-05 12:11 → Application still RUNNING, 72.1h duration, all 10 cores in use
Five Whys
Why #1: Why is the application deadlocked?
The Spark Connect server did not terminate after stage 44.0 was cancelled.
Stage was cancelled by the driver, executors killed tasks 249 and 250, but the SparkContext remained alive. In batch mode, a failed job kills the entire context. Spark Connect — a persistent long-lived session — is designed so that one failed query does not crash the server. However, here the session entered a state where resources are held but no new queries can proceed: the abort handler is stuck in an infinite retry loop, blocking the driver thread that should process the failure and release resources.
Evidence: Spark Master UI shows app-20260802111013-0002 as RUNNING with 72.1h duration. All 10 cores still allocated. Spark Connect port 4040 still responds on ASUS.
Why #2: Why didn't the abort handler complete?
Iceberg's abort handler entered a nested interruption deadlock between Spark cancellation and S3 cleanup.
The chain:
- Spark cancels the stage → sends
Thread.interrupt()to all task threads - Iceberg
DataAndMetadataWritingSparkTaskenters itsfinallyblock and attempts abort commit - Abort commit calls
S3FileIO.deleteFile()to clean up uncommitted staging files from S3 - AWS SDK
InterruptMonitor.checkInterrupted()detects the interrupt flag (still set from step 1) and throwsSdkInterruptedException - Iceberg
Tasks.java:413retry logic catches the exception and re-attempts the delete InterruptMonitoragain finds the flag → throws again- Infinite loop: retry → interrupt → retry — without ever clearing
Thread.interrupted()
Evidence: executor 3 stderr shows SdkInterruptedException at InterruptMonitor.checkInterrupted(), chained through S3FileIO.deleteFile() → RollingFileWriter.close() → DataAndMetadataWritingSparkTask.abort().
Why #3: Why wasn't the interrupt flag cleared before cleanup?
None of the three components implements the interrupted flag → clear → cleanup → re-interrupt pattern.
This is a gap at the intersection of three projects:
| Component | Expected behavior | Actual behavior |
|---|---|---|
| Spark | Cancel the task, allow cleanup to complete | Interrupts everything, including cleanup |
| Iceberg | Run cleanup in finally without interruption |
Calls S3 API without protecting against the interrupt flag |
| AWS SDK | Detect interruption and report | Throws exception but does NOT clear the flag |
The correct pattern — missing from Iceberg's RollingFileWriter.close() and S3FileIO.deleteFile() — is:
boolean wasInterrupted = Thread.interrupted(); // clear + check
try {
s3.deleteFile(path); // now without the flag
} finally {
if (wasInterrupted) Thread.currentThread().interrupt(); // restore
}
Evidence: stack trace directly shows the call path: DataAndMetadataWritingSparkTask.run() → abort() → RollingFileWriter.close() → S3FileIO.deleteFile() → InterruptMonitor.checkInterrupted(). No Thread.interrupted() call exists anywhere in this chain.
Why #4: Why did we enter the abort path in the first place?
The MERGE INTO failed with MERGE_CARDINALITY_VIOLATION — the source dataset contained duplicate rows matching the same ON key.
The ON search condition of the MERGE statement matched a single row
from the target table with multiple rows of the source table.
SQLSTATE: 23K01
The source CTE (fed to the USING clause of MERGE) contained multiple rows with identical merge key values, each matching the same target row. Iceberg (and the SQL standard) cannot determine which source row to apply — update, delete, or ignore. This is a data integrity error.
Evidence: executor 3 stderr: Stage cancelled: [MERGE_CARDINALITY_VIOLATION] The ON search condition of the MERGE statement matched a single row from the target table with multiple rows of the source table.
Why #5: Why did the source data contain duplicates? — ROOT CAUSE
The ETL pipeline lacks a defensive data quality gate between ingestion and MERGE.
Three specific gaps:
-
No deduplication before MERGE. The source CTE does not include
QUALIFY ROW_NUMBER() OVER (PARTITION BY <merge_keys> ORDER BY <criterion>) = 1or an equivalentGROUP BYto guarantee key uniqueness. The MERGE blindly trusts the input data. -
No pre-merge validation. There is no assertion like:
sql SELECT merge_key, COUNT(*) FROM source GROUP BY merge_key HAVING COUNT(*) > 1that would fail early with a clear error before reaching the MERGE. -
No idempotency protection at the orchestrator level. If the ingestion step was re-triggered (replaying data) or the upstream source sent duplicates, this is not detected until the MERGE itself fails — at which point the transaction is already in flight and cleanup is attempted.
Evidence: the error manifested at the Iceberg MERGE level (deep in the write path), not at the data ingestion level. No upstream check exists in the staging DAG code at src/entsoe_pipeline/lakehouse/core/merge_to_iseberg.py.
Summary
No deduplication / pre-merge validation in ETL pipeline
↓ therefore
Source data contains duplicate merge keys
↓ therefore
MERGE_CARDINALITY_VIOLATION — Iceberg rejects ambiguous MERGE
↓ therefore
Spark driver cancels Stage 44.0, fires Thread.interrupt()
↓ therefore
Iceberg abort handler hits S3 delete with interrupt flag still set
↓ therefore
Infinite retry loop: SdkInterruptedException → retry → SdkInterruptedException
↓ therefore
Spark Connect session deadlocked, 10 cores + 10 GiB lost for 72+ hours
Action Items
Immediate (hours — contain, restore service)
- [ ] Kill application
app-20260802111013-0002via Spark Master UI orspark-submit --kill— @stan_buren, due 2026-08-05 - [ ] Check for orphaned staging files in
s3://lakehouse-tables/staging/energyprices_12_1_d_r3/— @stan_buren, due 2026-08-05 - [ ] Verify cluster resources are released (cores/memory return to pool) — @stan_buren, due 2026-08-05
Short-Term (days — prevent this specific cause from recurring)
- [ ] Add
QUALIFY ROW_NUMBER() OVER (PARTITION BY <merge_keys> ORDER BY <criterion>) = 1in source CTE ofmerge_to_iseberg.py— @stan_buren, due 2026-08-07 - [ ] Add pre-merge assertion: if
COUNT(DISTINCT merge_key) != COUNT(*)in source → fail with clear error BEFORE MERGE — @stan_buren, due 2026-08-07
Long-Term (weeks — systemic process change)
- [ ] Replace MERGE with
INSERT OVERWRITEorCREATE OR REPLACEfor full partition rewrites — idempotent by nature in Iceberg — @stan_buren, due 2026-08-14 - [ ] Enable
spark.sql.adaptive.enabled=trueandspark.sql.adaptive.coalescePartitions.enabled=trueto reduce staging file count — @stan_buren, due 2026-08-14 - [ ] File upstream bug report with Apache Iceberg: abort path should clear interrupt flag before S3 cleanup — @stan_buren, due 2026-08-21
Lessons Learned
- Spark Connect sessions survive job failures by design — but that means a stuck abort handler blocks the entire cluster indefinitely. Every Spark Connect job with Iceberg writes needs a timeout + forced kill wrapper.
- MERGE_CARDINALITY_VIOLATION is a data quality problem, not a Spark problem. The fix is deduplication in the ETL, not changing Spark config.
Thread.interrupt()flag is sticky across libraries. Anyfinallyblock performing I/O after cancellation MUST clear and restore it. This is missing from Iceberg → AWS SDK integration.- Every MERGE INTO should be preceded by an assertion on source key uniqueness. Failing late (inside Iceberg's write path) means cleanup is attempted under cancellation — the worst possible context.
References
- Spark Master UI:
http://192.168.0.131:8080— applicationapp-20260802111013-0002 - Executor 3 stderr logs:
http://192.168.0.210:8081/logPage/?appId=app-20260802111013-0002&executorId=3&logType=stderr - MERGE implementation:
src/entsoe_pipeline/lakehouse/core/merge_to_iseberg.py - Staging DAG:
dags/staging.py