zomato-ai-data-engineering-pipeline

# zomato-ai-data-engineering-pipeline > Skill by [ara.so](https://ara.so) — Data Skills collection. Complete batch data engineering pipeline that processes food delivery data through a medallion architecture (Bronze → Silver → Gold) using Amazon S3, Snowflake, dbt, Airflow orchestration, and OpenAI-powered AI capabilities (LLM enrichment, RAG, text-to-SQL). ## Project Overview **Pipeline Flow:** ``` CSVs → S3 Data Lake → Snowflake RAW (Bronze) → dbt STAGING (Silver) → dbt MARTS (Gold) → AI Layer ``` **Architecture Layers:** - **Bronze (RAW)**: Direct `COPY INTO` from S3 via storage integration - **Silver (STAGING)**: dbt views for cleaning, typing, renaming - **Gold (MARTS)**: Dimensions, incremental facts (MERGE), business aggregates, SCD2 snapshots - **AI**: LLM enrichment, RAG chat, text-to-SQL queries **Data Scale:** - 10M orders - 23M order items - 300K text reviews - 7 source tables (restaurants, users, food, menu, orders, order_items, reviews) ## Installation & Setup ### Prerequisites ```bash # Clone and get dataset git clone https://github.com/darshilparmar/zomato-ai-data-engineering-end-to-end-project cd zomato-ai-data-engineering-end-to-end-project # Download CSVs from Google Drive (link in README) → place in data/ ``` ### AWS S3 Setup ```bash # 1. Create S3 bucket aws s3 mb s3://your-zomato-bucket # 2. Upload data to S3 aws s3 sync data/ s3://your-zomato-bucket/raw/ --exclude "*" \ --include "restaurants/*" \ --include "users/*" \ --include "food/*" \ --include "menu/*" \ --include "orders/*" \ --include "order_items/*" \ --include "reviews/*" # 3. Create IAM policy (use aws/iam/s3-read-policy.json) aws iam create-policy \ --policy-name zomato-s3-read \ --policy-document file://aws/iam/s3-read-policy.json # 4. Create IAM role with initial trust policy aws iam create-role \ --role-name snowflake-s3-role \ --assume-role-policy-document file://aws/iam/snowflake-role-trust-policy-initial.json # 5. Attach policy to role aws iam attach-role-policy \ --role-name snowflake-s3-role \ --policy-arn arn:aws:iam::YOUR_ACCOUNT:policy/zomato-s3-read ``` ### Snowflake Setup ```sql -- 1. Create warehouse, database, schemas CREATE WAREHOUSE ZOMATO_WH WITH WAREHOUSE_SIZE = 'MEDIUM' AUTO_SUSPEND = 60 AUTO_RESUME = TRUE; CREATE DATABASE ZOMATO; USE DATABASE ZOMATO; CREATE SCHEMA RAW; CREATE SCHEMA STAGING; CREATE SCHEMA MARTS; CREATE SCHEMA SNAPSHOTS; CREATE SCHEMA AI; -- 2. Create role and grant permissions CREATE ROLE DBT_ROLE; GRANT USAGE ON WAREHOUSE ZOMATO_WH TO ROLE DBT_ROLE; GRANT ALL ON DATABASE ZOMATO TO ROLE DBT_ROLE; GRANT ALL ON ALL SCHEMAS IN DATABASE ZOMATO TO ROLE DBT_ROLE; GRANT ROLE DBT_ROLE TO USER YOUR_USER; -- 3. Create storage integration (replace with your IAM role ARN) CREATE STORAGE INTEGRATION s3_zomato_integration TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = 'S3' ENABLED = TRUE STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::YOUR_ACCOUNT:role/snowflake-s3-role' STORAGE_ALLOWED_LOCATIONS = ('s3://your-zomato-bucket/raw/'); -- 4. Get Snowflake's IAM user ARN and external ID DESC STORAGE INTEGRATION s3_zomato_integration; -- Copy STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID -- 5. Update IAM role trust policy with these values (aws/iam/snowflake-role-trust-policy-final.json) -- 6. Create external stage CREATE STAGE s3_stage STORAGE_INTEGRATION = s3_zomato_integration URL = 's3://your-zomato-bucket/raw/'; -- 7. Create RAW tables CREATE TABLE RAW.RESTAURANTS ( restaurant_id NUMBER, name VARCHAR, city VARCHAR, rating FLOAT, rating_count NUMBER, cost VARCHAR, cuisine VARCHAR, lic_no VARCHAR, link VARCHAR, address VARCHAR, menu VARCHAR ); CREATE TABLE RAW.USERS ( user_id NUMBER, name VARCHAR, email VARCHAR, password VARCHAR, age NUMBER, gender VARCHAR, marital_status VARCHAR, occupation VARCHAR, monthly_income NUMBER, educational_qualifications VARCHAR, family_size NUMBER ); CREATE TABLE RAW.FOOD ( food_id NUMBER, item VARCHAR, veg_or_non_veg VARCHAR ); CREATE TABLE RAW.MENU ( menu_id NUMBER, restaurant_id NUMBER, food_id NUMBER, cuisine VARCHAR, price NUMBER ); CREATE TABLE RAW.ORDERS ( order_id NUMBER, user_id NUMBER, restaurant_id NUMBER, order_date DATE, order_time TIME, order_status VARCHAR, order_value NUMBER ); CREATE TABLE RAW.ORDER_ITEMS ( order_item_id NUMBER, order_id NUMBER, food_id NUMBER, quantity NUMBER, price NUMBER ); CREATE TABLE RAW.REVIEWS ( review_id NUMBER, order_id NUMBER, restaurant_id NUMBER, user_id NUMBER, rating NUMBER, review_text VARCHAR, review_date DATE ); ``` ### dbt Configuration ```bash cd zomato # Create profiles.yml (or update ~/.dbt/profiles.yml) cat > profiles.yml <<EOF zomato: target: dev outputs: dev: type: snowflake account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}" user: "{{ env_var('SNOWFLAKE_USER') }}" password: "{{ env_var('SNOWFLAKE_PASSWORD') }}" role: DBT_ROLE database: ZOMATO warehouse: ZOMATO_WH schema: STAGING threads: 4 EOF # Set environment variables export SNOWFLAKE_ACCOUNT=your_account.region export SNOWFLAKE_USER=your_user export SNOWFLAKE_PASSWORD=your_password # Test connection dbt debug # Install dependencies dbt deps ``` ### Airflow Setup ```bash cd airflow # Create .env from example cp example.env .env # Edit .env with your credentials # SNOWFLAKE_ACCOUNT=your_account.region # SNOWFLAKE_USER=your_user # SNOWFLAKE_PASSWORD=your_password # SNOWFLAKE_ROLE=DBT_ROLE # SNOWFLAKE_WAREHOUSE=ZOMATO_WH # SNOWFLAKE_DATABASE=ZOMATO # OPENAI_API_KEY=your_openai_key # S3_BUCKET=your-zomato-bucket # SAMPLE_N=1000 # Build and start Airflow docker compose build docker compose up -d # Access Airflow UI # http://localhost:8080 (admin/admin) ``` ## Key dbt Models ### Staging (Silver Layer) ```yaml # models/staging/schema.yml version: 2 sources: - name: raw database: ZOMATO schema: RAW tables: - name: restaurants - name: users - name: food - name: menu - name: orders - name: order_items - name: reviews ``` ```sql -- models/staging/stg_restaurants.sql WITH source AS ( SELECT * FROM {{ source('raw', 'restaurants') }} ), cleaned AS ( SELECT restaurant_id, TRIM(name) AS restaurant_name, LOWER(TRIM(city)) AS city, rating, rating_count, -- Parse cost: '₹ 200' → 200, '--' → NULL TRY_CAST( REPLACE(REPLACE(cost, '₹', ''), ' ', '') AS NUMBER ) AS avg_cost_for_two, TRIM(cuisine) AS cuisine, NULLIF(TRIM(lic_no), '--') AS license_number, link AS restaurant_url, address, menu AS menu_url FROM source ) SELECT * FROM cleaned ``` ```sql -- models/staging/stg_orders.sql WITH source AS ( SELECT * FROM {{ source('raw', 'orders') }} ), cleaned AS ( SELECT order_id, user_id, restaurant_id, order_date, order_time, LOWER(TRIM(order_status)) AS order_status, order_value, -- Derive delivery flag CASE WHEN order_status = 'delivered' THEN TRUE ELSE FALSE END AS is_delivered, -- Derive cancellation flag CASE WHEN order_status IN ('cancelled', 'canceled') THEN TRUE ELSE FALSE END AS is_cancelled FROM source ) SELECT * FROM cleaned ``` ### Marts (Gold Layer) ```sql -- models/marts/dim_restaurants.sql {{ config( materialized='table' ) }} SELECT restaurant_id, restaurant_name, city, rating, rating_count, avg_cost_for_two, cuisine, license_number, restaurant_url, address FROM {{ ref('stg_restaurants') }} ``` ```sql -- models/marts/dim_customer.sql {{ config( materialized='table' ) }} WITH customers AS ( SELECT user_id, name AS customer_name, LOWER(email) AS email, age, gender, marital_status, occupation, monthly_income, educational_qualifications, family_size, -- Age segmentation CASE WHEN age < 25 THEN '18-24' WHEN age BETWEEN 25 AND 34 THEN '25-34' WHEN age BETWEEN 35 AND 44 THEN '35-44' WHEN age BETWEEN 45 AND 54 THEN '45-54' WHEN age >= 55 THEN '55+' ELSE 'Unknown' END AS age_segment FROM {{ ref('stg_users') }} ) SELECT * FROM customers ``` ```sql -- models/marts/fct_orders.sql {{ config( materialized='incremental', unique_key='order_id', on_schema_change='append_new_columns' ) }} WITH orders AS ( SELECT order_id, user_id, restaurant_id, order_date, order_time, order_status, order_value, is_delivered, is_cancelled FROM {{ ref('stg_orders') }} {% if is_incremental() %} WHERE order_date > (SELECT MAX(order_date) FROM {{ this }}) {% endif %} ) SELECT * FROM orders ``` ```sql -- models/marts/fact_order_items.sql {{ config( materialized='incremental', unique_key='order_item_id', on_schema_change='append_new_columns' ) }} WITH order_items AS ( SELECT oi.order_item_id, oi.order_id, oi.food_id, oi.quantity, oi.price, oi.quantity * oi.price AS line_total, o.order_date FROM {{ ref('stg_order_items') }} oi JOIN {{ ref('stg_orders') }} o ON oi.order_id = o.order_id {% if is_incremental() %} WHERE o.order_date > (SELECT MAX(order_date) FROM {{ this }}) {% endif %} ) SELECT * FROM order_items ``` ```sql -- models/marts/mart_daily_city_revenue.sql {{ config( materialized='table' ) }} WITH daily_metrics AS ( SELECT o.order_date, r.city, COUNT(DISTINCT o.order_id) AS total_orders, COUNT(DISTINCT CASE WHEN o.is_delivered THEN o.order_id END) AS delivered_orders, COUNT(DISTINCT CASE WHEN o.is_cancelled THEN o.order_id END) AS cancelled_orders, SUM(CASE WHEN o.is_delivered THEN o.order_value ELSE 0 END) AS gmv, AVG(CASE WHEN o.is_delivered THEN o.order_value END) AS aov, COUNT(DISTINCT o.user_id) AS active_customers, COUNT(DISTINCT o.restaurant_id) AS active_restaurants FROM {{ ref('fct_orders') }} o JOIN {{ ref('dim_restaurants') }} r ON o.restaurant_id = r.restaurant_id GROUP BY o.order_date, r.city ) SELECT order_date, city, total_orders, delivered_orders, cancelled_orders, ROUND(cancelled_orders::FLOAT / NULLIF(total_orders, 0) * 100, 2) AS cancellation_rate_pct, gmv, aov, active_customers, active_restaurants, ROUND(gmv / NULLIF(active_restaurants, 0), 2) AS revenue_per_restaurant FROM daily_metrics ``` ### dbt Testing ```yaml # models/marts/schema.yml version: 2 models: - name: dim_restaurants description: Restaurant dimension columns: - name: restaurant_id description: Primary key tests: - unique - not_null - name: city tests: - not_null - name: fct_orders description: Orders fact table (incremental) columns: - name: order_id description: Primary key tests: - unique - not_null - name: user_id tests: - not_null - relationships: to: ref('dim_customer') field: user_id - name: restaurant_id tests: - not_null - relationships: to: ref('dim_restaurants') field: restaurant_id - name: order_status tests: - accepted_values: values: ['delivered', 'cancelled', 'pending', 'preparing'] ``` ## Airflow DAG