# murari123-byte/dataflow-processing-engine architecture

> Architecture diagram of the GitHub repository murari123-byte/dataflow-processing-engine, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-11-14). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/murari123-byte/dataflow-processing-engine
- Repository: https://github.com/murari123-byte/dataflow-processing-engine

## Overview

DataFlow Processing Engine is a PySpark data pipeline. Its main workflow reads customer and order CSVs, validates and separates accepted/rejected records, enriches orders and aggregates daily sales, builds fact/dimension datasets, reconciles counts, and publishes Parquet outputs while recording run status. Separate capabilities include reproducible data generation, analytics over published data, and aggregation/join benchmarks. Edges follow the sampled pipeline calls and explicit README descriptions; analytics-to-output access is documented, but its implementation details were not sampled.

Main entry points: [`src/run_pipeline.py`](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_pipeline.py) (Pipeline orchestrator), [`src/data/publisher.py`](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/data/publisher.py) (Staged Parquet publisher), [`src/processing/validation.py`](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/validation.py) (Validation and quarantine), [`src/run_benchmark.py`](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_benchmark.py) (Aggregation benchmark)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_inputs["Data inputs"]
  node_customers_csv["Customer CSV"]
  node_orders_csv["Order CSV"]
  node_generator["Dataset generator<br/>[generate_data.py]"]
end

subgraph group_pipeline["Processing pipeline"]
  node_config["Runtime configuration<br/>[config.py]"]
  node_pipeline_runner["Pipeline orchestrator<br/>[run_pipeline.py]"]
  node_validation["Validation and quarantine<br/>[validation.py]"]
  node_optimized_transform["Enrichment and aggregation<br/>[optimized.py]"]
  node_dimensional_model["Fact and dimensions"]
  node_reconciliation["Reconciliation checks<br/>[reconciliation.py]"]
end

subgraph group_outputs["Published data"]
  node_publisher["Staged Parquet publisher<br/>[publisher.py]"]
  node_valid_orders[("Valid orders")]
  node_daily_sales[("Daily sales")]
  node_rejected_orders[("Rejected orders")]
  node_dimensional_outputs[("Fact and dimensions")]
  node_reconciliation_report["Reconciliation report"]
  node_run_log["Pipeline run log"]
end

subgraph group_operations["Operations and analysis"]
  node_monitoring["Run monitoring<br/>[monitoring.py]"]
  node_analytics["SQL analytics<br/>[analytics.py]"]
  node_analytics_runner["Analytics entry point<br/>[run_analytics.py]"]
end

subgraph group_performance["Performance benchmarks"]
  node_baseline["Baseline transform<br/>[baseline.py]"]
  node_benchmark_runner["Aggregation benchmark<br/>[run_benchmark.py]"]
  node_join_benchmark["Join benchmark"]
end

node_operator(("Pipeline operator"))

node_operator -->|"starts"| node_pipeline_runner
node_generator -->|"generates"| node_customers_csv
node_generator -->|"generates"| node_orders_csv
node_pipeline_runner -->|"loads settings"| node_config
node_pipeline_runner -->|"reads"| node_orders_csv
node_pipeline_runner -->|"reads"| node_customers_csv
node_pipeline_runner -->|"validates"| node_validation
node_validation -->|"returns accepted"| node_valid_orders
node_validation -->|"returns rejected"| node_rejected_orders
node_pipeline_runner -->|"transforms valid orders"| node_optimized_transform
node_optimized_transform -->|"aggregates"| node_daily_sales
node_pipeline_runner -->|"builds model"| node_dimensional_model
node_dimensional_model -->|"returns fact and dimensions"| node_dimensional_outputs
node_pipeline_runner -->|"reconciles"| node_reconciliation
node_reconciliation -->|"checks totals"| node_daily_sales
node_pipeline_runner -->|"writes report"| node_reconciliation_report
node_pipeline_runner -->|"publishes datasets"| node_publisher
node_publisher -->|"writes Parquet"| node_valid_orders
node_publisher -->|"writes Parquet"| node_daily_sales
node_publisher -->|"writes Parquet"| node_rejected_orders
node_publisher -->|"writes Parquet"| node_dimensional_outputs
node_pipeline_runner -->|"records run"| node_monitoring
node_monitoring -->|"appends record"| node_run_log
node_analytics_runner -->|"runs queries"| node_analytics
node_analytics -->|"queries published data"| node_valid_orders
node_benchmark_runner -->|"validates inputs"| node_validation
node_benchmark_runner -->|"compares transforms"| node_baseline
node_benchmark_runner -->|"compares transforms"| node_optimized_transform
node_operator -->|"runs benchmark"| node_join_benchmark

click node_generator "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/generator/generate_data.py"
click node_config "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/config.py"
click node_pipeline_runner "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_pipeline.py"
click node_validation "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/validation.py"
click node_optimized_transform "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/optimized.py"
click node_dimensional_model "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/dimensional_model.py"
click node_reconciliation "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/reconciliation.py"
click node_publisher "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/data/publisher.py"
click node_monitoring "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/monitoring.py"
click node_analytics "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/analytics.py"
click node_analytics_runner "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_analytics.py"
click node_baseline "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/baseline.py"
click node_benchmark_runner "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_benchmark.py"
click node_join_benchmark "https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_join_benchmark.py"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_customers_csv,node_orders_csv,node_generator toneBlue
class node_config,node_pipeline_runner,node_validation,node_optimized_transform,node_dimensional_model,node_reconciliation toneAmber
class node_publisher,node_valid_orders,node_daily_sales,node_rejected_orders,node_dimensional_outputs,node_reconciliation_report,node_run_log toneMint
class node_monitoring,node_analytics,node_analytics_runner toneRose
class node_baseline,node_benchmark_runner,node_join_benchmark,node_operator toneIndigo
```

## Components

### Data inputs

- **Customer CSV**
- **Order CSV**
- [Dataset generator](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/generator/generate_data.py): `src/generator/generate_data.py`

### Processing pipeline

- [Runtime configuration](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/config.py): `src/config.py`
- [Pipeline orchestrator](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_pipeline.py): `src/run_pipeline.py`
- [Validation and quarantine](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/validation.py): `src/processing/validation.py`
- [Enrichment and aggregation](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/optimized.py): `src/processing/optimized.py`
- [Fact and dimensions](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/dimensional_model.py): `src/processing/dimensional_model.py`
- [Reconciliation checks](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/reconciliation.py): `src/processing/reconciliation.py`

### Published data

- [Staged Parquet publisher](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/data/publisher.py): `src/data/publisher.py`
- **Valid orders**
- **Daily sales**
- **Rejected orders**
- **Fact and dimensions**
- **Reconciliation report**
- **Pipeline run log**

### Operations and analysis

- [Run monitoring](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/monitoring.py): `src/monitoring.py`
- [SQL analytics](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/analytics.py): `src/processing/analytics.py`
- [Analytics entry point](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_analytics.py): `src/run_analytics.py`

### Performance benchmarks

- [Baseline transform](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/processing/baseline.py): `src/processing/baseline.py`
- [Aggregation benchmark](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_benchmark.py): `src/run_benchmark.py`
- [Join benchmark](https://github.com/murari123-byte/dataflow-processing-engine/blob/main/src/run_join_benchmark.py): `src/run_join_benchmark.py`

### Other components

- **Pipeline operator**

## Connections

- Pipeline operator → Pipeline orchestrator: starts (evidence: `README.md`)
- Dataset generator → Customer CSV: generates (evidence: `README.md`)
- Dataset generator → Order CSV: generates (evidence: `README.md`)
- Pipeline orchestrator → Runtime configuration: loads settings (evidence: `src/run_pipeline.py`)
- Pipeline orchestrator → Order CSV: reads (evidence: `src/run_pipeline.py`)
- Pipeline orchestrator → Customer CSV: reads (evidence: `src/run_pipeline.py`)
- Pipeline orchestrator → Validation and quarantine: validates (evidence: `src/run_pipeline.py`)
- Validation and quarantine → Valid orders: returns accepted (evidence: `src/processing/validation.py`)
- Validation and quarantine → Rejected orders: returns rejected (evidence: `src/processing/validation.py`)
- Pipeline orchestrator → Enrichment and aggregation: transforms valid orders (evidence: `src/run_pipeline.py`)
- Enrichment and aggregation → Daily sales: aggregates (evidence: `src/processing/optimized.py`)
- Pipeline orchestrator → Fact and dimensions: builds model (evidence: `src/run_pipeline.py`)
- Fact and dimensions → Fact and dimensions: returns fact and dimensions (evidence: `src/processing/dimensional_model.py`)
- Pipeline orchestrator → Reconciliation checks: reconciles (evidence: `src/run_pipeline.py`)
- Reconciliation checks → Daily sales: checks totals (evidence: `src/processing/reconciliation.py`)
- Pipeline orchestrator → Reconciliation report: writes report (evidence: `src/run_pipeline.py`)
- Pipeline orchestrator → Staged Parquet publisher: publishes datasets (evidence: `src/run_pipeline.py`)
- Staged Parquet publisher → Valid orders: writes Parquet (evidence: `src/data/publisher.py`)
- Staged Parquet publisher → Daily sales: writes Parquet (evidence: `src/data/publisher.py`)
- Staged Parquet publisher → Rejected orders: writes Parquet (evidence: `src/data/publisher.py`)
- Staged Parquet publisher → Fact and dimensions: writes Parquet (evidence: `src/data/publisher.py`)
- Pipeline orchestrator → Run monitoring: records run (evidence: `src/run_pipeline.py`)
- Run monitoring → Pipeline run log: appends record (evidence: `src/monitoring.py`)
- Analytics entry point → SQL analytics: runs queries (evidence: `src/run_analytics.py`)
- SQL analytics → Valid orders: queries published data (evidence: `README.md`)
- Aggregation benchmark → Validation and quarantine: validates inputs (evidence: `src/run_benchmark.py`)
- Aggregation benchmark → Baseline transform: compares transforms (evidence: `src/run_benchmark.py`)
- Aggregation benchmark → Enrichment and aggregation: compares transforms (evidence: `src/run_benchmark.py`)
- Pipeline operator → Join benchmark: runs benchmark (evidence: `README.md`)