# udaradev/fleet-lambda-platform architecture

> Architecture diagram of the GitHub repository udaradev/fleet-lambda-platform, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-09-28). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/udaradev/fleet-lambda-platform
- Repository: https://github.com/udaradev/fleet-lambda-platform

## Overview

Fleet Lambda Platform simulates ride-hailing telemetry and expenses, producing near-real-time fleet insights and authoritative daily profitability reports. Telemetry is published to Kafka and consumed independently by a speed stream for live serving and a raw stream for immutable archival. Airflow triggers batch reconciliation of archived trips with expense files; reconciliation validates and aggregates inputs, then updates daily serving data and publishes integrity-manifested reports. The API exposes live metrics, alerts, health, and reports. The README documents monitoring and data-quality workflows beyond the sampled source; those branches are included at the documented level, without assuming their internal task wiring.

Main entry points: [`batch/reconcile.py`](https://github.com/udaradev/fleet-lambda-platform/blob/main/batch/reconcile.py) (Batch reconciliation), [`streaming/speed_job.py`](https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/speed_job.py) (Speed stream), [`airflow/dags/data_quality_dag.py`](https://github.com/udaradev/fleet-lambda-platform/blob/main/airflow/dags/data_quality_dag.py) (Quality workflow), [`api/main.py`](https://github.com/udaradev/fleet-lambda-platform/blob/main/api/main.py) (FastAPI service)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_ingestion["Simulation and ingestion"]
  node_telemetry["Telemetry producer<br/>[producer_stream.py]"]
  node_expense["Expense producer<br/>[producer_batch.py]"]
  node_kafka["Kafka events"]
  node_expense_files[("Expense CSV")]
end

subgraph group_streaming["Lambda streaming"]
  node_speed["Speed stream<br/>[speed_job.py]"]
  node_raw["Raw stream<br/>[raw_job.py]"]
  node_validation["Event validation<br/>[validation.py]"]
  node_sink["Serving sink<br/>[sink.py]"]
  node_staged_sink["Staged merge<br/>[staged_sink.py]"]
  node_minio[("MinIO archive")]
end

subgraph group_batch["Batch and quality"]
  node_archive["Archive verification<br/>[archive.py]"]
  node_airflow_daily["Daily workflow"]
  node_reconcile["Batch reconciliation<br/>[reconcile.py]"]
  node_spark_batch["Trip aggregation<br/>[spark_reconcile.py]"]
  node_domain["Business rules<br/>[domain.py]"]
  node_quality["Quality workflow"]
end

subgraph group_serving["Serving and reporting"]
  node_postgres[("PostgreSQL serving")]
  node_reports[("Report exports")]
  node_api["FastAPI service<br/>[main.py]"]
end

subgraph group_observability["Monitoring"]
  node_prometheus["Prometheus"]
  node_grafana["Grafana"]
end

node_business_user(("Business user"))

node_telemetry -->|"publishes events"| node_kafka
node_expense -->|"writes expenses"| node_expense_files
node_kafka -->|"consumed by"| node_speed
node_kafka -->|"consumed by"| node_raw
node_speed -->|"validates events"| node_validation
node_speed -->|"writes serving data"| node_sink
node_speed -->|"merges batches"| node_staged_sink
node_sink -->|"updates live tables"| node_postgres
node_staged_sink -->|"merges staged data"| node_postgres
node_raw -->|"archives Parquet"| node_minio
node_airflow_daily -.->|"runs pipeline"| node_reconcile
node_reconcile -->|"reads expenses"| node_expense_files
node_reconcile -->|"verifies archive"| node_archive
node_archive -->|"checks objects"| node_minio
node_reconcile -->|"aggregates trips"| node_spark_batch
node_spark_batch -->|"reads Parquet"| node_minio
node_reconcile -->|"applies rules"| node_domain
node_reconcile -->|"stores daily results"| node_postgres
node_reconcile -->|"publishes exports"| node_reports
node_quality -.->|"checks data quality"| node_postgres
node_quality -.->|"checks integrity"| node_archive
node_quality -.->|"checks manifests"| node_reports
node_api -->|"queries serving data"| node_postgres
node_api -->|"serves metrics and reports"| node_business_user
node_prometheus -.->|"scrapes metrics"| node_api
node_grafana -.->|"queries metrics"| node_prometheus

click node_telemetry "https://github.com/udaradev/fleet-lambda-platform/blob/main/simulators/producer_stream.py"
click node_expense "https://github.com/udaradev/fleet-lambda-platform/blob/main/simulators/producer_batch.py"
click node_speed "https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/speed_job.py"
click node_raw "https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/raw_job.py"
click node_validation "https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/validation.py"
click node_sink "https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/sink.py"
click node_staged_sink "https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/staged_sink.py"
click node_archive "https://github.com/udaradev/fleet-lambda-platform/blob/main/common/archive.py"
click node_airflow_daily "https://github.com/udaradev/fleet-lambda-platform/blob/main/airflow/dags/daily_profitability_dag.py"
click node_reconcile "https://github.com/udaradev/fleet-lambda-platform/blob/main/batch/reconcile.py"
click node_spark_batch "https://github.com/udaradev/fleet-lambda-platform/blob/main/batch/spark_reconcile.py"
click node_domain "https://github.com/udaradev/fleet-lambda-platform/blob/main/common/domain.py"
click node_quality "https://github.com/udaradev/fleet-lambda-platform/blob/main/airflow/dags/data_quality_dag.py"
click node_api "https://github.com/udaradev/fleet-lambda-platform/blob/main/api/main.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_telemetry,node_expense,node_kafka,node_expense_files,node_business_user toneBlue
class node_speed,node_raw,node_validation,node_sink,node_staged_sink,node_minio toneAmber
class node_archive,node_airflow_daily,node_reconcile,node_spark_batch,node_domain,node_quality toneMint
class node_postgres,node_reports,node_api toneRose
class node_prometheus,node_grafana toneIndigo
```

## Components

### Simulation and ingestion

- [Telemetry producer](https://github.com/udaradev/fleet-lambda-platform/blob/main/simulators/producer_stream.py): `simulators/producer_stream.py`
- [Expense producer](https://github.com/udaradev/fleet-lambda-platform/blob/main/simulators/producer_batch.py): `simulators/producer_batch.py`
- **Kafka events**
- **Expense CSV**

### Lambda streaming

- [Speed stream](https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/speed_job.py): `streaming/speed_job.py`
- [Raw stream](https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/raw_job.py): `streaming/raw_job.py`
- [Event validation](https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/validation.py): `streaming/validation.py`
- [Serving sink](https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/sink.py): `streaming/sink.py`
- [Staged merge](https://github.com/udaradev/fleet-lambda-platform/blob/main/streaming/staged_sink.py): `streaming/staged_sink.py`
- **MinIO archive**

### Batch and quality

- [Archive verification](https://github.com/udaradev/fleet-lambda-platform/blob/main/common/archive.py): `common/archive.py`
- [Daily workflow](https://github.com/udaradev/fleet-lambda-platform/blob/main/airflow/dags/daily_profitability_dag.py): `airflow/dags/daily_profitability_dag.py`
- [Batch reconciliation](https://github.com/udaradev/fleet-lambda-platform/blob/main/batch/reconcile.py): `batch/reconcile.py`
- [Trip aggregation](https://github.com/udaradev/fleet-lambda-platform/blob/main/batch/spark_reconcile.py): `batch/spark_reconcile.py`
- [Business rules](https://github.com/udaradev/fleet-lambda-platform/blob/main/common/domain.py): `common/domain.py`
- [Quality workflow](https://github.com/udaradev/fleet-lambda-platform/blob/main/airflow/dags/data_quality_dag.py): `airflow/dags/data_quality_dag.py`

### Serving and reporting

- **PostgreSQL serving**
- **Report exports**
- [FastAPI service](https://github.com/udaradev/fleet-lambda-platform/blob/main/api/main.py): `api/main.py`

### Monitoring

- **Prometheus**
- **Grafana**

### Other components

- **Business user**

## Connections

- Telemetry producer → Kafka events: publishes events
- Expense producer → Expense CSV: writes expenses
- Kafka events → Speed stream: consumed by
- Kafka events → Raw stream: consumed by
- Speed stream → Event validation: validates events
- Speed stream → Serving sink: writes serving data
- Speed stream → Staged merge: merges batches
- Serving sink → PostgreSQL serving: updates live tables
- Staged merge → PostgreSQL serving: merges staged data
- Raw stream → MinIO archive: archives Parquet
- Daily workflow → Batch reconciliation: runs pipeline
- Batch reconciliation → Expense CSV: reads expenses
- Batch reconciliation → Archive verification: verifies archive
- Archive verification → MinIO archive: checks objects
- Batch reconciliation → Trip aggregation: aggregates trips
- Trip aggregation → MinIO archive: reads Parquet
- Batch reconciliation → Business rules: applies rules
- Batch reconciliation → PostgreSQL serving: stores daily results
- Batch reconciliation → Report exports: publishes exports
- Quality workflow → PostgreSQL serving: checks data quality
- Quality workflow → Archive verification: checks integrity
- Quality workflow → Report exports: checks manifests
- FastAPI service → PostgreSQL serving: queries serving data
- FastAPI service → Business user: serves metrics and reports
- Prometheus → FastAPI service: scrapes metrics
- Grafana → Prometheus: queries metrics