# trannhatnguyen2/nyc_taxi_data_pipeline architecture

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

- Interactive diagram: https://gitdiagram.com/trannhatnguyen2/nyc_taxi_data_pipeline
- Repository: https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline

## Overview

NYC Taxi Data Pipeline ingests TLC trip files through a scheduled batch ELT flow, stages processed data in PostgreSQL, validates it, and builds a warehouse star schema with dbt. A separate CDC/streaming path sends PostgreSQL changes through Debezium and Kafka into MinIO; the repository also includes streaming-read code and Trino configuration. The batch code and Airflow DAG were excerpted; CDC, dbt, validation, and Trino connections are included only where the README documents them.

Main entry points: [`airflow/dags/elt_pipeline_dag.py`](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/airflow/dags/elt_pipeline_dag.py) (Airflow DAG), [`batch_processing/datalake_to_dw.py`](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/batch_processing/datalake_to_dw.py) (Batch staging load), [`scripts/extract_load.py`](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/extract_load.py) (File extraction), [`scripts/transform_data.py`](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/transform_data.py) (Trip normalization)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_batch["Batch ingestion"]
  node_airflow["Airflow DAG"]
  node_extract["File extraction<br/>[extract_load.py]"]
  node_transform["Trip normalization<br/>[transform_data.py]"]
  node_delta["Delta conversion"]
  node_batchload["Batch staging load<br/>[datalake_to_dw.py]"]
end

subgraph group_warehouse["Warehouse modeling"]
  node_validation["Data validation"]
  node_dbt["Star schema models"]
  node_dbtmacros["Warehouse descriptions"]
end

subgraph group_stream["CDC streaming"]
  node_cdcconfig["CDC connector setup<br/>[taxi-nyc-cdc.json]"]
  node_cdcscript["Connector registration<br/>[run.sh]"]
  node_streamdb["Database event producer"]
  node_streamjson["JSON event producer"]
  node_streamload["Stream lake writer"]
  node_streamread["Parquet stream reader"]
end

subgraph group_platform["Shared data services"]
  node_minio[("MinIO data lake")]
  node_postgres[("PostgreSQL")]
  node_kafka["Kafka topics"]
  node_trino["Trino query engine"]
end

node_operator(("Pipeline operator"))

node_operator -->|"starts"| node_airflow
node_airflow -->|"calls"| node_extract
node_airflow -->|"calls"| node_transform
node_airflow -->|"calls"| node_delta
node_extract -->|"uploads raw"| node_minio
node_transform -->|"writes processed"| node_minio
node_delta -->|"reads and writes"| node_minio
node_batchload -->|"reads parquet"| node_minio
node_batchload -->|"writes staging"| node_postgres
node_postgres -->|"staging validated"| node_validation
node_postgres -->|"staged data modeled"| node_dbt
node_dbt -.->|"uses descriptions"| node_dbtmacros
node_operator -->|"registers connector"| node_cdcscript
node_cdcscript -.->|"uses configuration"| node_cdcconfig
node_postgres -->|"changes captured"| node_kafka
node_streamdb -->|"inserts records"| node_postgres
node_streamjson -.->|"produces events"| node_kafka
node_kafka -->|"streams events"| node_streamload
node_streamload -->|"writes parquet"| node_minio
node_minio -.->|"provides parquet"| node_streamread

click node_airflow "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/airflow/dags/elt_pipeline_dag.py"
click node_extract "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/extract_load.py"
click node_transform "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/transform_data.py"
click node_delta "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/convert_to_delta.py"
click node_batchload "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/batch_processing/datalake_to_dw.py"
click node_validation "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/tree/main/data_validation/gx"
click node_dbt "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/tree/main/dbt_nyc/models/production"
click node_cdcconfig "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/debezium/configs/taxi-nyc-cdc.json"
click node_cdcscript "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/debezium/run.sh"
click node_streamdb "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/utils/streaming_data_db.py"
click node_streamjson "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/utils/streaming_data_json.py"
click node_streamload "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/stream_processing/streaming_to_datalake.py"
click node_streamread "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/stream_processing/read_parquet_streaming.py"
click node_trino "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/trino/catalog/datalake.properties"
click node_dbtmacros "https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/tree/main/dbt_nyc/macros"

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_airflow,node_extract,node_transform,node_delta,node_batchload toneBlue
class node_validation,node_dbt,node_dbtmacros toneAmber
class node_cdcconfig,node_cdcscript,node_streamdb,node_streamjson,node_streamload,node_streamread toneMint
class node_minio,node_postgres,node_kafka,node_trino toneRose
class node_operator toneIndigo
```

## Components

### Batch ingestion

- [Airflow DAG](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/airflow/dags/elt_pipeline_dag.py): `airflow/dags/elt_pipeline_dag.py`
- [File extraction](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/extract_load.py): `scripts/extract_load.py`
- [Trip normalization](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/transform_data.py): `scripts/transform_data.py`
- [Delta conversion](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/scripts/convert_to_delta.py): `scripts/convert_to_delta.py`
- [Batch staging load](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/batch_processing/datalake_to_dw.py): `batch_processing/datalake_to_dw.py`

### Warehouse modeling

- [Data validation](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/tree/main/data_validation/gx): `data_validation/gx`
- [Star schema models](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/tree/main/dbt_nyc/models/production): `dbt_nyc/models/production`
- [Warehouse descriptions](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/tree/main/dbt_nyc/macros): `dbt_nyc/macros`

### CDC streaming

- [CDC connector setup](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/debezium/configs/taxi-nyc-cdc.json): `debezium/configs/taxi-nyc-cdc.json`
- [Connector registration](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/debezium/run.sh): `debezium/run.sh`
- [Database event producer](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/utils/streaming_data_db.py): `utils/streaming_data_db.py`
- [JSON event producer](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/utils/streaming_data_json.py): `utils/streaming_data_json.py`
- [Stream lake writer](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/stream_processing/streaming_to_datalake.py): `stream_processing/streaming_to_datalake.py`
- [Parquet stream reader](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/stream_processing/read_parquet_streaming.py): `stream_processing/read_parquet_streaming.py`

### Shared data services

- **MinIO data lake**
- **PostgreSQL**
- **Kafka topics**
- [Trino query engine](https://github.com/trannhatnguyen2/nyc_taxi_data_pipeline/blob/main/trino/catalog/datalake.properties): `trino/catalog/datalake.properties`

### Other components

- **Pipeline operator**

## Connections

- Pipeline operator → Airflow DAG: starts (evidence: `README.md`)
- Airflow DAG → File extraction: calls (evidence: `airflow/dags/elt_pipeline_dag.py`)
- Airflow DAG → Trip normalization: calls (evidence: `airflow/dags/elt_pipeline_dag.py`)
- Airflow DAG → Delta conversion: calls (evidence: `airflow/dags/elt_pipeline_dag.py`)
- File extraction → MinIO data lake: uploads raw (evidence: `scripts/extract_load.py`)
- Trip normalization → MinIO data lake: writes processed (evidence: `scripts/transform_data.py`)
- Delta conversion → MinIO data lake: reads and writes (evidence: `scripts/convert_to_delta.py`)
- Batch staging load → MinIO data lake: reads parquet (evidence: `batch_processing/datalake_to_dw.py`)
- Batch staging load → PostgreSQL: writes staging (evidence: `batch_processing/datalake_to_dw.py`)
- PostgreSQL → Data validation: staging validated (evidence: `README.md`)
- PostgreSQL → Star schema models: staged data modeled (evidence: `README.md`)
- Star schema models → Warehouse descriptions: uses descriptions
- Pipeline operator → Connector registration: registers connector (evidence: `README.md`)
- Connector registration → CDC connector setup: uses configuration
- PostgreSQL → Kafka topics: changes captured (evidence: `README.md`)
- Database event producer → PostgreSQL: inserts records (evidence: `utils/streaming_data_db.py`)
- JSON event producer → Kafka topics: produces events
- Kafka topics → Stream lake writer: streams events (evidence: `stream_processing/streaming_to_datalake.py`)
- Stream lake writer → MinIO data lake: writes parquet (evidence: `stream_processing/streaming_to_datalake.py`)
- MinIO data lake → Parquet stream reader: provides parquet