uber-ai-data-pipeline

KO | EN

gitlite — search

#javascript #cross-platform #claude-code #dotnet #ai-agents #mysql #postgresql #svg #pdf #github

uber-ai-data-pipeline

(★ 9)

No description available.

파일 탐색기

ZIP 다운로드

- enrich_reviews.py
- rag_chat.py
- review_embeddings.praquet
- text_to_sql.py
- - uber_batch.py
- docker-compose.yaml
- Dockerfile
- Airflow.png
- Architecture.png
- S3_Custom_Policy.png
- Streamlit_rag_chat.png
- Streamlit_text_to_sql.png
- uber_batch_graph.png
- 01_uber_pipeline_setup.sql
- 02_uber_storage_intergration.sql
- 03_uber_stage_and_formats.sql
- 04_uber_raw_table.sql
- 05_uber_copy_into.sql
- - .gitkeep
- - .gitkeep
  - generate_schema_name.sql
- - .gitkeep
- - .gitkeep
- - .gitkeep
- .gitignore
- dbt_project.yml
- README.md
.gitignore
README.md
requirements.txt

# CDN으로 사용하기

jsDelivr

jsDelivr는 공개 GitHub 리포지토리를 별도 설정 없이 CDN으로 즉시 서빙합니다. 버전과 파일을 고르면 웹페이지에 바로 붙일 수 있는 링크와 예시 코드가 만들어집니다.

버전

파일

링크

예시

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

An end-to-end, production-style data engineering pipeline for a food-delivery business (Uber Eats), built on a modern data stack and extended with an AI layer for review enrichment, retrieval-augmented chat, and natural-language-to-SQL analytics.

Raw CSVs land in Amazon S3, get copied into Snowflake, are modeled through a medallion architecture with dbt (staging → marts), enriched with an LLM, and orchestrated end-to-end as a daily Airflow DAG running in Docker.

## Architecture

![Architecture](https://raw.githubusercontent.com/saviya98/uber-ai-data-pipeline/main/docs/Architecture.png)

The pipeline follows a bronze / silver / gold pattern:

- **Bronze (`RAW`)** — CSVs copied verbatim into Snowflake from an S3 stage.
- **Silver (`STAGING`)** — dbt views that clean, type-cast, and standardize the raw data (e.g. parsing `city` out of a free-text address, coercing `rating`/`cost` strings to numbers, deriving `is_delivered`).
- **Gold (`MARTS`)** — a star schema plus purpose-built analytics tables that answer specific business questions, refreshed incrementally.
- **AI (`AI`)** — an LLM-enriched schema layered on top of the gold layer, feeding both a dbt model and two Streamlit apps.

## Tech stack

| Layer | Tools |
|---|---|
| Storage / ingestion | Amazon S3, Snowflake `COPY INTO` |
| Transformation | dbt (`dbt-snowflake`) |
| Orchestration | Apache Airflow on Docker Compose (LocalExecutor) |
| AI / LLM | OpenAI `gpt-4o-mini` (classification, text-to-SQL, RAG) and `text-embedding-3-small` (semantic search) |
| Application layer | Streamlit |
| Languages | SQL, Python |

## Data model

**Sources** (`UBER.RAW`): `restaurants`, `users`, `food`, `menu`, `orders`, `order_items`, `reviews`

**Staging** (`UBER.STAGING`, views): one `stg_*` model per source table — deduplicates keys, normalizes city names, casts strings pulled from messy source columns (e.g. `try_to_decimal`/`regexp_substr` on `rating`, `cost`, `age`), and derives `is_delivered` from `order_status`.

**Marts** (`UBER.MARTS`, tables):

| Model | Grain | Notes |
|---|---|---|
| `dim_customer` | 1 row / customer | Derives an `age_segment` (Gen Z / Millennial / Gen X / Boomer / Unknown) |
| `dim_restaurants` | 1 row / restaurant | City, cuisine, rating, cost for two |
| `dim_food` | 1 row / food item | Name, veg/non-veg flag |
| `dim_date` | 1 row / day | Generated date spine, 2024-01-01 → 2026-12-31 |
| `fact_orders` | 1 row / order | **Incremental** (merge on `order_id`, watermarked on `order_timestamp`) |
| `fact_order_items` | 1 row / order line | **Incremental** (merge on `order_item_id`) |
| `mart_daily_city_revenue` | day × city | GMV, cancel rate, average order value |
| `mart_restaurant_performance` | restaurant | Orders, revenue, avg rating, avg delivery time |
| `mart_delivery_sla` | city × hour | p50 / p90 delivery time, delivered-order counts |
| `mart_review_insights` | city × topic × sentiment | Blends dbt-modeled reviews with LLM-enriched sentiment/topic (tagged `ai`, built in its own DAG step) |

**AI** (`UBER.AI`): `review_enriched` — one row per review, written by `ai/enrich_reviews.py`.

## Data quality

dbt tests enforce the contract at every layer: `unique` / `not_null` on all primary keys (`stg_orders`, `stg_order_items`, `stg_restaurants`, `stg_users`, `stg_reviews`, `fact_orders`), a `relationships` test asserting every `fact_orders.customer_id` exists in `dim_customer`, and an `accepted_values` test constraining `order_status` to `Delivered` / `Cancelled` / `Refunded`. Run them with:

```bash
dbt build --project-dir uber_pipeline --profiles-dir uber_pipeline
```

## The AI layer

1. **Review enrichment** (`ai/enrich_reviews.py`) — pulls unclassified rows from `RAW.REVIEWS`, sends each comment to `gpt-4o-mini` with a structured-JSON prompt, and writes `sentiment_label`, `sentiment_score`, `topic`, and `key_issue` back to `AI.REVIEW_ENRICHED`. Idempotent: only reviews not already present in the enriched table are processed.
2. **RAG chat** (`ai/rag_chat.py`) — a Streamlit app that embeds a sample of reviews with `text-embedding-3-small`, caches the embeddings to Parquet, retrieves the top-k most similar reviews by cosine similarity for a user's question, and answers grounded strictly in that retrieved context.
3. **Text-to-SQL** (`ai/text_to_sql.py`) — a Streamlit app that turns a plain-English question into a single `SELECT` statement against a documented mart schema, guards the generated SQL with a `SELECT`/`WITH`-only check plus a blocklist of DDL/DML keywords before executing it against Snowflake, and auto-charts two-column numeric results.

Streamlit RAG Chat

![Rag_Chat](https://raw.githubusercontent.com/saviya98/uber-ai-data-pipeline/main/docs/Streamlit_rag_chat.png)

Streamlit Text-to-SQL

![Text-to-SQL](https://raw.githubusercontent.com/saviya98/uber-ai-data-pipeline/main/docs/Streamlit_text_to_sql.png)

## Orchestration

`airflow/dags/uber_batch.py` defines a daily DAG with a strict dependency chain:

```
reload_raw  →  dbt_build_core  →  enrich_reviews  →  dbt_build_ai
```

- `reload_raw` — `COPY INTO` each `RAW` table from the S3 stage.
- `dbt_build_core` — `dbt build --exclude tag:ai` (staging + core marts).
- `enrich_reviews` — runs `ai/enrich_reviews.py` against the freshly loaded reviews.
- `dbt_build_ai` — `dbt build --select tag:ai` (currently `mart_review_insights`), which depends on the enrichment step having just run.

Airflow runs as Airflow 3.x under Docker Compose (`airflow/docker-compose.yaml`) with Postgres as the metadata DB, a dedicated `dbt-snowflake` virtualenv baked into the image so dbt's dependencies never collide with Airflow's, and the Snowflake/OpenAI credentials injected as environment variables.

![Orchestration](https://raw.githubusercontent.com/saviya98/uber-ai-data-pipeline/main/docs/Airflow.png)

## Repository structure

```
uber_data_pipeline/
├── ai/                       # LLM enrichment + AI-powered Streamlit apps
│   ├── enrich_reviews.py     # Review sentiment/topic classification
│   ├── rag_chat.py           # RAG chat over reviews
│   └── text_to_sql.py        # Natural-language → SQL analytics assistant
├── airflow/                  # Orchestration
│   ├── dags/uber_batch.py    # Daily DAG: raw load → dbt → enrich → dbt(ai)
│   ├── docker-compose.yaml
│   └── Dockerfile
├── uber_pipeline/            # dbt project
│   ├── models/
│   │   ├── staging/          # 1:1 cleaned views on top of RAW sources
│   │   └── marts/            # Star schema + analytics marts
│   ├── macros/
│   ├── snapshots/
│   ├── seeds/
│   └── dbt_project.yml
├── docs/                     # Architecture references (e.g. S3 IAM policy)
├── data/                     # Raw CSVs (not committed)
└── README.md
```

## Running it locally

**Prerequisites:** a Snowflake account/warehouse, an AWS S3 bucket staged as a Snowflake external stage, an OpenAI API key, and Docker.

```bash
# 1. dbt
cd uber_pipeline
dbt deps
dbt build            # requires uber_pipeline/profiles.yml (gitignored — configure your own)

# 2. AI apps (from repo root)
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt   # openai, snowflake-connector-python, streamlit, python-dotenv, pandas, numpy
streamlit run ai/rag_chat.py
streamlit run ai/text_to_sql.py

# 3. Orchestration
cd airflow
cp .env.example .env              # fill in Snowflake + OpenAI credentials
docker compose build && docker compose up -d
# Airflow UI: http://localhost:8080 (admin/admin)
```

Required environment variables (kept out of version control via `.gitignore`):

| Variable | Used by |
|---|---|
| `SNOWFLAKE_ACCOUNT`, `SNOWFLAKE_USER`, `SNOWFLAKE_PASSWORD`, `SNOWFLAKE_WAREHOUSE`, `SNOWFLAKE_DATABASE`, `SNOWFLAKE_SCHEMA`, `SNOWFLAKE_ROLE` | dbt, Airflow's Snowflake connection, `ai/*.py` |
| `OPENAI_API_KEY` | `ai/enrich_reviews.py`, `ai/rag_chat.py`, `ai/text_to_sql.py` |
| `SAMPLE_N` | Batch size for review enrichment |

## Skills demonstrated

- Cloud ingestion and warehousing: S3 → Snowflake staged loads
- Dimensional modeling: medallion architecture (bronze/silver/gold) and a star schema (dims + incremental facts)
- dbt: sources, staging/mart layering, schema tests, referential-integrity tests, incremental models with `merge` strategy, model tagging (`tag:ai`) to control selective builds
- Orchestration: multi-step Airflow DAG mixing SQL, `dbt build`, and Python tasks with explicit dependencies; containerized with Docker Compose
- Applied AI engineering: structured-output prompting for classification, embeddings + cosine similarity for retrieval (RAG), guardrailed natural-language-to-SQL generation
- Python application development: Streamlit UIs backed by a live warehouse connection

## Data source

Built on a Kaggle food-delivery dataset (Zomato/Uber Eats-style: users, restaurants, food, menu, orders, order items, reviews), used here for learning and portfolio purposes.

uber-ai-data-pipeline

파일 탐색기

# CDN으로 사용하기

Was this content helpful?

Recent Feedback

Download README