# nguyenlehoaitam/topic120-distributeddatabase architecture

> Architecture diagram of the GitHub repository nguyenlehoaitam/topic120-distributeddatabase, 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/nguyenlehoaitam/topic120-distributeddatabase
- Repository: https://github.com/nguyenlehoaitam/topic120-distributeddatabase

## Overview

This is a simulated three-stage weather-event streaming and alert pipeline, with batch runners for 30- and 60-second windows and an interactive dashboard. The primary workflow reads generated sensor CSV events, cleans and transforms them in separate processes, then applies event-time lateness handling and county-window aggregation before reporting results. Checkpointing and a crash demonstration are supporting capabilities. The runner and node wiring is documented and partly sampled; dashboard and failure-demo details rely on the README and partial excerpts.

Main entry points: [`src/main_30s.py`](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/main_30s.py) (30s pipeline runner), [`src/main_60s.py`](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/main_60s.py) (60s pipeline runner), [`src/failure_sim.py`](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/failure_sim.py) (Crash simulation), [`src/node_aggregate.py`](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/node_aggregate.py) (County aggregation)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_ingest["Data input"]
  node_generator["Dataset generator<br/>[generator.py]"]
  node_csv["Weather events CSV"]
end

subgraph group_pipeline["Stream processing"]
  node_runner30["30s pipeline runner<br/>[main_30s.py]"]
  node_runner60["60s pipeline runner<br/>[main_60s.py]"]
  node_clean["Clean and deduplicate<br/>[node_clean.py]"]
  node_transform["Unit transform<br/>[node_transform.py]"]
  node_aggregate["County aggregation<br/>[node_aggregate.py]"]
  node_windows["Tumbling windows<br/>[windowing.py]"]
  node_watermark["Late-event policy<br/>[watermark.py]"]
end

subgraph group_state["State and recovery"]
  node_state_mgr["Checkpoint manager<br/>[state_manager.py]"]
  node_checkpoints[("Node checkpoints")]
  node_failure["Crash simulation<br/>[failure_sim.py]"]
end

subgraph group_reporting["Reporting and dashboard"]
  node_report30["30s report<br/>[report_30s.txt]"]
  node_report60["60s report<br/>[report_60s.txt]"]
  node_dashboard["Interactive dashboard<br/>[dashboard.py]"]
  node_metrics["Latency metrics<br/>[metrics.py]"]
  node_failurelog["Failure log<br/>[failure_log.txt]"]
end

node_operator(("Operator"))

node_operator -->|"starts generation"| node_generator
node_generator -->|"writes events"| node_csv
node_operator -->|"starts run"| node_runner30
node_operator -->|"starts run"| node_runner60
node_runner30 -->|"reads events"| node_csv
node_runner60 -->|"reads events"| node_csv
node_runner30 -->|"dispatches events"| node_clean
node_runner60 -->|"dispatches events"| node_clean
node_clean -->|"queues clean events"| node_transform
node_transform -->|"queues transformed events"| node_aggregate
node_aggregate -->|"adds events"| node_windows
node_aggregate -->|"checks lateness"| node_watermark
node_clean -->|"saves and loads"| node_state_mgr
node_transform -->|"saves and loads"| node_state_mgr
node_aggregate -->|"saves and loads"| node_state_mgr
node_state_mgr -->|"reads and writes"| node_checkpoints
node_runner30 -->|"writes report"| node_report30
node_runner60 -->|"writes report"| node_report60
node_operator -->|"sets parameters"| node_dashboard
node_dashboard -->|"reads events"| node_csv
node_dashboard -->|"aggregates events"| node_windows
node_dashboard -->|"applies lateness"| node_watermark
node_runner30 -.->|"records timings"| node_metrics
node_runner60 -.->|"records timings"| node_metrics
node_operator -->|"starts demo"| node_failure
node_failure -->|"starts node"| node_clean
node_failure -->|"simulates crash"| node_transform
node_failure -->|"starts resilient node"| node_aggregate
node_failure -->|"writes log"| node_failurelog

click node_generator "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/generator.py"
click node_csv "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/data/weather_sensors.csv"
click node_runner30 "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/main_30s.py"
click node_runner60 "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/main_60s.py"
click node_clean "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/node_clean.py"
click node_transform "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/node_transform.py"
click node_aggregate "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/node_aggregate.py"
click node_windows "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/windowing.py"
click node_watermark "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/watermark.py"
click node_state_mgr "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/state_manager.py"
click node_failure "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/failure_sim.py"
click node_report30 "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/results/report_30s.txt"
click node_report60 "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/results/report_60s.txt"
click node_dashboard "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/dashboard.py"
click node_metrics "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/metrics.py"
click node_failurelog "https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/results/failure_log.txt"

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_generator,node_csv toneBlue
class node_runner30,node_runner60,node_clean,node_transform,node_aggregate,node_windows,node_watermark toneAmber
class node_state_mgr,node_checkpoints,node_failure toneMint
class node_report30,node_report60,node_dashboard,node_metrics,node_failurelog toneRose
class node_operator toneIndigo
```

## Components

### Data input

- [Dataset generator](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/generator.py): `src/generator.py`
- [Weather events CSV](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/data/weather_sensors.csv): `data/weather_sensors.csv`

### Stream processing

- [30s pipeline runner](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/main_30s.py): `src/main_30s.py`
- [60s pipeline runner](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/main_60s.py): `src/main_60s.py`
- [Clean and deduplicate](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/node_clean.py): `src/node_clean.py`
- [Unit transform](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/node_transform.py): `src/node_transform.py`
- [County aggregation](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/node_aggregate.py): `src/node_aggregate.py`
- [Tumbling windows](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/windowing.py): `src/windowing.py`
- [Late-event policy](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/watermark.py): `src/watermark.py`

### State and recovery

- [Checkpoint manager](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/state_manager.py): `src/state_manager.py`
- **Node checkpoints**
- [Crash simulation](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/failure_sim.py): `src/failure_sim.py`

### Reporting and dashboard

- [30s report](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/results/report_30s.txt): `results/report_30s.txt`
- [60s report](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/results/report_60s.txt): `results/report_60s.txt`
- [Interactive dashboard](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/dashboard.py): `dashboard.py`
- [Latency metrics](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/src/metrics.py): `src/metrics.py`
- [Failure log](https://github.com/nguyenlehoaitam/topic120-distributeddatabase/blob/main/results/failure_log.txt): `results/failure_log.txt`

### Other components

- **Operator**

## Connections

- Operator → Dataset generator: starts generation
- Dataset generator → Weather events CSV: writes events
- Operator → 30s pipeline runner: starts run
- Operator → 60s pipeline runner: starts run
- 30s pipeline runner → Weather events CSV: reads events
- 60s pipeline runner → Weather events CSV: reads events
- 30s pipeline runner → Clean and deduplicate: dispatches events
- 60s pipeline runner → Clean and deduplicate: dispatches events
- Clean and deduplicate → Unit transform: queues clean events
- Unit transform → County aggregation: queues transformed events
- County aggregation → Tumbling windows: adds events
- County aggregation → Late-event policy: checks lateness
- Clean and deduplicate → Checkpoint manager: saves and loads
- Unit transform → Checkpoint manager: saves and loads
- County aggregation → Checkpoint manager: saves and loads
- Checkpoint manager → Node checkpoints: reads and writes
- 30s pipeline runner → 30s report: writes report
- 60s pipeline runner → 60s report: writes report
- Operator → Interactive dashboard: sets parameters
- Interactive dashboard → Weather events CSV: reads events
- Interactive dashboard → Tumbling windows: aggregates events
- Interactive dashboard → Late-event policy: applies lateness
- 30s pipeline runner → Latency metrics: records timings
- 60s pipeline runner → Latency metrics: records timings
- Operator → Crash simulation: starts demo
- Crash simulation → Clean and deduplicate: starts node
- Crash simulation → Unit transform: simulates crash
- Crash simulation → County aggregation: starts resilient node
- Crash simulation → Failure log: writes log