بناء «Data Lakehouse» باستخدام «DuckDB» و«dbt»

# Clone the repository
git clone https://github.com/AlharbiAbdullah/data-lakehouse
cd data-lakehouse

# Install dependencies with UV
uv sync

# Or with pip
pip install -r requirements.txt

data-lakehouse/
|-- data/
|   |-- raw/                    # Parquet files (gitignored)
|   +-- warehouse/              # DuckDB database
|
|-- dbt_project/
|   |-- models/
|   |   |-- staging/            # Bronze layer
|   |   |-- intermediate/       # Silver layer
|   |   +-- marts/              # Gold layer
|   |-- macros/                 # Reusable SQL
|   +-- seeds/                  # Reference data
|
|-- scripts/
|   |-- download_data.py        # Fetch NYC Taxi data
|   |-- setup_warehouse.py      # Initialize DuckDB
|   +-- run_pipeline.py         # Orchestrate everything
|
+-- .github/workflows/
    +-- dbt_ci.yml              # CI/CD pipeline

+==============================================================================+
|                            MEDALLION LAYERS                                  |
+==============================================================================+
|                                                                              |
|    +--------------+       +------------------+       +--------------+        |
|    |    BRONZE    |       |      SILVER      |       |     GOLD     |        |
|    |   (staging)  | ----> |  (intermediate)  | ----> |    (marts)   |        |
|    +--------------+       +------------------+       +--------------+        |
|                                                                              |
|     Raw ingestion          Cleaned &                  Analytics-ready        |
|     Type casting           Validated                  Aggregated             |
|     Hash-based IDs         Zone enrichment            Incremental            |
|                            Calculated fields          Daily metrics          |
|                                                                              |
+==============================================================================+

# Download all data
uv run python scripts/download_data.py

# Set up DuckDB warehouse
uv run python scripts/setup_warehouse.py

{{
    config(
        materialized='view'
    )
}}

with source as (
    select * from {{ source('raw', 'yellow_tripdata') }}
),

with_base_hash as (
    select
        -- Deterministic ID from key fields
        {{ generate_trip_id(
            'tpep_pickup_datetime',
            'tpep_dropoff_datetime',
            'PULocationID',
            'DOLocationID',
            'fare_amount',
            'trip_distance',
            'passenger_count'
        ) }} as base_hash,

        -- Trip type identifier
        'yellow' as trip_type,

        -- Timestamps
        tpep_pickup_datetime as pickup_datetime,
        tpep_dropoff_datetime as dropoff_datetime,

        -- Locations
        PULocationID as pickup_zone_id,
        DOLocationID as dropoff_zone_id,

        -- Trip details
        cast(passenger_count as integer) as passenger_count,
        cast(trip_distance as double) as trip_distance,

        -- Fare components
        cast(fare_amount as double) as fare_amount,
        cast(tip_amount as double) as tip_amount,
        cast(total_amount as double) as total_amount,

        -- Metadata
        current_timestamp as loaded_at

    from source
    where tpep_pickup_datetime is not null
      and tpep_dropoff_datetime is not null
),

staged as (
    select
        -- Unique trip ID: base_hash + row number for duplicates
        base_hash || '_' || cast(row_number() over (
            partition by base_hash order by pickup_datetime
        ) as varchar) as trip_id,
        * exclude (base_hash)
    from with_base_hash
)

select * from staged

{% macro generate_trip_id(pickup_datetime, dropoff_datetime, pickup_zone,
                          dropoff_zone, extra_field_1, extra_field_2, extra_field_3) %}
    md5(
        coalesce(cast({{ pickup_datetime }} as varchar), '') ||
        '|' ||
        coalesce(cast({{ dropoff_datetime }} as varchar), '') ||
        '|' ||
        coalesce(cast({{ pickup_zone }} as varchar), '') ||
        '|' ||
        coalesce(cast({{ dropoff_zone }} as varchar), '') ||
        '|' ||
        coalesce(cast({{ extra_field_1 }} as varchar), '') ||
        '|' ||
        coalesce(cast({{ extra_field_2 }} as varchar), '') ||
        '|' ||
        coalesce(cast({{ extra_field_3 }} as varchar), '')
    )
{% endmacro %}

{{
    config(
        materialized='view'
    )
}}

with trips as (
    select * from {{ ref('int_trips_unioned') }}
),

zones as (
    select * from {{ ref('stg_taxi_zones') }}
),

enriched as (
    select
        t.trip_id,
        t.trip_type,
        t.pickup_datetime,
        t.dropoff_datetime,
        t.pickup_zone_id,
        t.dropoff_zone_id,
        t.passenger_count,
        t.trip_distance,
        t.fare_amount,
        t.tip_amount,
        t.total_amount,
        t.loaded_at,

        -- Pickup zone info
        pz.zone_name as pickup_zone_name,
        pz.borough as pickup_borough,

        -- Dropoff zone info
        dz.zone_name as dropoff_zone_name,
        dz.borough as dropoff_borough,

        -- Calculated fields
        datediff('minute', t.pickup_datetime, t.dropoff_datetime)
            as trip_duration_minutes,

        -- Average speed (mph)
        case
            when t.trip_distance > 0
                 and datediff('minute', t.pickup_datetime, t.dropoff_datetime) > 0
            then t.trip_distance / (datediff('minute', t.pickup_datetime,
                                             t.dropoff_datetime) / 60.0)
            else null
        end as avg_speed_mph,

        -- Tip percentage
        case
            when t.fare_amount > 0 and t.tip_amount is not null
            then (t.tip_amount / t.fare_amount) * 100
            else null
        end as tip_percentage

    from trips t
    left join zones pz on t.pickup_zone_id = pz.zone_id
    left join zones dz on t.dropoff_zone_id = dz.zone_id
)

select * from enriched

{{
    config(
        materialized='incremental',
        unique_key=['trip_date', 'pickup_borough', 'trip_type'],
        incremental_strategy='merge'
    )
}}

with trips as (
    select * from {{ ref('int_trips_validated') }}

    {% if is_incremental() %}
    -- 3-day lookback for late-arriving data
    where pickup_datetime >= (
        select dateadd('day', -3, max(trip_date))
        from {{ this }}
    )
    {% endif %}
),

daily_aggregates as (
    select
        cast(date_trunc('day', pickup_datetime) as date) as trip_date,
        pickup_borough,
        trip_type,

        -- Trip counts
        count(*) as total_trips,
        sum(coalesce(passenger_count, 0)) as total_passengers,

        -- Distance metrics
        sum(coalesce(trip_distance, 0)) as total_distance_miles,
        avg(trip_distance) as avg_distance_miles,

        -- Fare metrics
        sum(coalesce(fare_amount, 0)) as total_fare,
        avg(fare_amount) as avg_fare,
        sum(coalesce(tip_amount, 0)) as total_tips,
        avg(tip_percentage) as avg_tip_percentage,

        -- Duration metrics
        avg(trip_duration_minutes) as avg_duration_minutes,

        -- Metadata
        current_timestamp as updated_at

    from trips
    where pickup_borough is not null
    group by 1, 2, 3
)

select * from daily_aggregates

version: 2

models:
  - name: stg_yellow_trips
    columns:
      - name: trip_id
        tests:
          - unique
          - not_null
      - name: pickup_datetime
        tests:
          - not_null
      - name: dropoff_datetime
        tests:
          - not_null
      - name: trip_type
        tests:
          - accepted_values:
              values: ['yellow']

cd dbt_project
uv run dbt test

name: dbt CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  dbt-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install UV
        run: curl -LsSf https://astral.sh/uv/install.sh | sh

      - name: Install dependencies
        run: uv sync

      - name: Download January data (CI subset)
        run: uv run python scripts/download_data.py --months 1

      - name: Setup warehouse
        run: uv run python scripts/setup_warehouse.py

      - name: Run dbt
        run: |
          cd dbt_project
          uv run dbt deps
          uv run dbt seed
          uv run dbt run
          uv run dbt test

# Full pipeline
uv run python scripts/run_pipeline.py

# Or step by step:
uv run python scripts/download_data.py   # ~10GB download
uv run python scripts/setup_warehouse.py

cd dbt_project
uv run dbt deps     # Install dbt packages
uv run dbt seed     # Load reference data
uv run dbt run      # Execute transformations
uv run dbt test     # Validate data quality

-- Daily trip summary by borough
SELECT
    trip_date,
    pickup_borough,
    trip_type,
    total_trips,
    avg_fare,
    avg_duration_minutes
FROM marts.fct_daily_trips
WHERE trip_date >= '2024-01-01'
ORDER BY trip_date, total_trips DESC;

-- Busiest zones
SELECT
    z.zone_name,
    z.borough,
    SUM(m.pickups) as total_pickups
FROM marts.fct_zone_metrics m
JOIN marts.dim_zones z ON m.zone_id = z.zone_id
GROUP BY z.zone_name, z.borough
ORDER BY total_pickups DESC
LIMIT 10;