# nalexx6/diploma architecture

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

- Interactive diagram: https://gitdiagram.com/nalexx6/diploma
- Repository: https://github.com/nalexx6/diploma

## Overview

This repository implements a NYC taxi trip analytics pipeline: trips can be simulated into Kafka, consumed and normalized by Spark streaming into Parquet, then aggregated by a Spark batch job and written to PostgreSQL. An EMR step manager launches and monitors the Spark jobs. A separate pandas application performs local trip analyses. The README is empty, so capabilities and wiring here are based on the sampled source and file tree; configuration details and unsampled infrastructure behavior are not inferred.

Main entry points: [`spark_apps/batch.py`](https://github.com/nalexx6/diploma/blob/master/spark_apps/batch.py) (Spark batch job), [`spark_apps/streaming.py`](https://github.com/nalexx6/diploma/blob/master/spark_apps/streaming.py) (Spark streaming job), [`spark_apps/simulation.py`](https://github.com/nalexx6/diploma/blob/master/spark_apps/simulation.py) (Trip simulator), [`spark_apps/utils/__init__.py`](https://github.com/nalexx6/diploma/blob/master/spark_apps/utils/__init__.py) (Spark utilities)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_orchestration["Job orchestration"]
  node_step_manager["EMR step manager<br/>[steps_manager.py]"]
end

subgraph group_ingestion["Trip ingestion"]
  node_stream_job["Spark streaming job<br/>[streaming.py]"]
  node_simulation["Trip simulator<br/>[simulation.py]"]
  node_kafka["Kafka topics"]
end

subgraph group_analytics["Trip analytics"]
  node_batch_job["Spark batch job<br/>[batch.py]"]
  node_spark_utils["Spark utilities<br/>[__init__.py]"]
  node_metrics["Trip metrics<br/>[metrics.py]"]
  node_config["Spark configuration<br/>[config.yaml]"]
  node_metadata["Taxi metadata"]
  node_pandas_app["Local analysis app<br/>[main.py]"]
  node_pandas_functions["Pandas analyses<br/>[functions.py]"]
end

subgraph group_outputs["Data outputs"]
  node_parquet[("Trip Parquet data")]
  node_postgres[("PostgreSQL metrics")]
  node_local_parquet[("Local Parquet files")]
end

node_operator(("Operator"))
node_emr["Amazon EMR"]

node_operator -->|"starts jobs"| node_step_manager
node_step_manager -->|"submits and monitors"| node_emr
node_emr -->|"runs submitted job"| node_batch_job
node_emr -->|"runs submitted job"| node_stream_job
node_simulation -->|"loads config and session"| node_spark_utils
node_simulation -->|"reads trips"| node_parquet
node_simulation -->|"publishes trips"| node_kafka
node_stream_job -->|"loads config and session"| node_spark_utils
node_stream_job -->|"reads trip events"| node_kafka
node_stream_job -->|"appends trip data"| node_parquet
node_batch_job -->|"loads data and saves metrics"| node_spark_utils
node_batch_job -->|"calls aggregations"| node_metrics
node_batch_job -->|"reads trip data"| node_parquet
node_batch_job -->|"reads licenses and zones"| node_metadata
node_pandas_app -->|"reads trip files"| node_local_parquet
node_pandas_app -->|"calls analyses"| node_pandas_functions
node_spark_utils -->|"writes metric tables"| node_postgres
node_batch_job -->|"loads configuration"| node_config
node_stream_job -->|"loads configuration"| node_config
node_simulation -->|"loads configuration"| node_config

click node_step_manager "https://github.com/nalexx6/diploma/blob/master/emr_steps_execution/steps_manager.py"
click node_batch_job "https://github.com/nalexx6/diploma/blob/master/spark_apps/batch.py"
click node_stream_job "https://github.com/nalexx6/diploma/blob/master/spark_apps/streaming.py"
click node_simulation "https://github.com/nalexx6/diploma/blob/master/spark_apps/simulation.py"
click node_spark_utils "https://github.com/nalexx6/diploma/blob/master/spark_apps/utils/__init__.py"
click node_metrics "https://github.com/nalexx6/diploma/blob/master/spark_apps/metrics.py"
click node_config "https://github.com/nalexx6/diploma/blob/master/spark_apps/config.yaml"
click node_metadata "https://github.com/nalexx6/diploma/tree/master/nyc_taxi_metadata"
click node_pandas_app "https://github.com/nalexx6/diploma/blob/master/python_apps/main.py"
click node_pandas_functions "https://github.com/nalexx6/diploma/blob/master/python_apps/functions.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_step_manager toneBlue
class node_stream_job,node_simulation,node_kafka toneAmber
class node_batch_job,node_spark_utils,node_metrics,node_config,node_metadata,node_pandas_app,node_pandas_functions toneMint
class node_parquet,node_postgres,node_local_parquet toneRose
class node_operator,node_emr toneIndigo
```

## Components

### Job orchestration

- [EMR step manager](https://github.com/nalexx6/diploma/blob/master/emr_steps_execution/steps_manager.py): `emr_steps_execution/steps_manager.py`

### Trip ingestion

- [Spark streaming job](https://github.com/nalexx6/diploma/blob/master/spark_apps/streaming.py): `spark_apps/streaming.py`
- [Trip simulator](https://github.com/nalexx6/diploma/blob/master/spark_apps/simulation.py): `spark_apps/simulation.py`
- **Kafka topics**

### Trip analytics

- [Spark batch job](https://github.com/nalexx6/diploma/blob/master/spark_apps/batch.py): `spark_apps/batch.py`
- [Spark utilities](https://github.com/nalexx6/diploma/blob/master/spark_apps/utils/__init__.py): `spark_apps/utils/__init__.py`
- [Trip metrics](https://github.com/nalexx6/diploma/blob/master/spark_apps/metrics.py): `spark_apps/metrics.py`
- [Spark configuration](https://github.com/nalexx6/diploma/blob/master/spark_apps/config.yaml): `spark_apps/config.yaml`
- [Taxi metadata](https://github.com/nalexx6/diploma/tree/master/nyc_taxi_metadata): `nyc_taxi_metadata`
- [Local analysis app](https://github.com/nalexx6/diploma/blob/master/python_apps/main.py): `python_apps/main.py`
- [Pandas analyses](https://github.com/nalexx6/diploma/blob/master/python_apps/functions.py): `python_apps/functions.py`

### Data outputs

- **Trip Parquet data**
- **PostgreSQL metrics**
- **Local Parquet files**

### Other components

- **Operator**
- **Amazon EMR**

## Connections

- Operator → EMR step manager: starts jobs
- EMR step manager → Amazon EMR: submits and monitors
- Amazon EMR → Spark batch job: runs submitted job
- Amazon EMR → Spark streaming job: runs submitted job
- Trip simulator → Spark utilities: loads config and session
- Trip simulator → Trip Parquet data: reads trips
- Trip simulator → Kafka topics: publishes trips
- Spark streaming job → Spark utilities: loads config and session
- Spark streaming job → Kafka topics: reads trip events
- Spark streaming job → Trip Parquet data: appends trip data
- Spark batch job → Spark utilities: loads data and saves metrics
- Spark batch job → Trip metrics: calls aggregations
- Spark batch job → Trip Parquet data: reads trip data
- Spark batch job → Taxi metadata: reads licenses and zones
- Local analysis app → Local Parquet files: reads trip files
- Local analysis app → Pandas analyses: calls analyses
- Spark utilities → PostgreSQL metrics: writes metric tables
- Spark batch job → Spark configuration: loads configuration
- Spark streaming job → Spark configuration: loads configuration
- Trip simulator → Spark configuration: loads configuration