Get configuration value.

Get configuration value by nested keys.

Get bucket name for a specific layer.

Get configured logger.

Standardized error handling across all processors. Args: error: The exception that occurred context:

Push metrics to Pushgateway with optional grouping key. Args: grouping_key: Dictionary of labels to use as grouping key.

Upsert DataFrame to Delta table using MERGE operation. Idempotent - safe to re-run. Args: table_name: Na

Get storage options for Delta Lake.

Upload JSON data to S3.

Shared metric tracking logic.

Calculate and write customer segmentation.

Calculate and write product performance.

Create configured DuckDB connection.

Deduplicate records against existing Delta table using partition pruning. Returns: (arrow_new, duplicates_removed)

Execute multiple aggregations in parallel using ThreadPoolExecutor. Args: aggregations: List of (name, function)

Load and validate configuration.

Load only today's partition for incremental aggregation.

Recursively replace ${VAR} with environment variables.

Execute Soda Core checks on the DataFrame. Returns a result dict for process().

Validate processor-specific configuration. Must be implemented by subclasses.

Write Arrow Table to Delta Lake using APPEND mode. Idempotent - safe to re-run.

Create Gold layer business metrics. Uses ThreadPoolExecutor for parallel table processing. Returns: {

Generate customer dimension.

Generate product catalog.

Generate incremental sales transactions with realistic data quality issues.

Ingest incremental sales to Bronze layer. Returns: {"key": s3_key, "count": record_count, "status": "success"}

Load product and customer dimensions (idempotent). Returns: {"status": "loaded" | "skipped", "products": int, "c

Load dimension tables (products and customers). Idempotent - skips if already loaded. Returns: {"status"

Check if object exists in S3.

Run Soda Core quality checks on Silver data. Returns: { "status": "success" | "failed" | "skippe

Process Bronze data to Silver Delta table. Args: bronze_key: S3 key of the Bronze data file Ret

Create daily business metrics as Delta tables. Processes all aggregations in parallel using ThreadPoolExecutor. Retu

Ingest sales data to Bronze layer. Returns: { "key": "s3_key_path", "count": rec

Process Bronze → Silver with cleaning and deduplication. Args: bronze_result: Output from ingest_sales_data task

Sales data pipeline with Bronze-Silver-Gold architecture. Uses Interface Inheritance pattern for clean, maintainable code.

Placeholder task for when aggregation is skipped. Uses AirflowSkipException for proper task state.

Track data quality failures.

Track number of records processed.

Track Delta table size.

Run Soda Core quality checks on Silver data. Returns: { "status": "success" | "failed" | "skippe

Context manager entry.

Context manager exit.

Validate quality checker configuration.

Validate Silver-specific configuration.

Validate Gold-specific configuration.

Validate Bronze-specific configuration.

Main processing logic - must be implemented by subclasses. Returns: Dict with processing results (status, counts

Determine if daily aggregation should run. Runs on first execution OR at configured hour (default: midnight UTC). Re

Upload Parquet data to S3.

Validate payment method.

Validate customer segment.

Validate transaction status.

Ensure total_amount matches quantity * unit_price.