С чего начинается дизайн

Прежде чем писать код, нужно ответить на пять вопросов:

1. Что моделируем? Какой бизнес-процесс? Какие сущности? Где их грейн (gran of measurement)?
2. Как партиционируем? По какому полю разбиваем данные на куски?
3. Как обеспечиваем идемпотентность? Где ключи, где MERGE, где детерминированные пути?
4. Как тестируем? Какие инварианты должны выполняться?
5. Как мониторим? Какие метрики собирать?

Это дизайн-документ, который ты обычно пишешь до кода. На production это PR-описание или RFC. В pet-project — это твоя заметка перед началом работы. На собеседовании — это рассказ “как я подходил к проекту”.

Поедем по очереди для NYC Taxi сценария.

1. Что моделируем: грейн

Грейн (grain) — это уровень детализации фактовой таблицы: что одна строка означает в реальном мире.

Для NYC Taxi грейн — одна поездка (one trip = one row). Каждая строка фактовой таблицы — это одна поездка такси с пикапом и дроп-офф-ом, fare, tip, distance.

Не одна “минута поездки”, не одна “оплата”, не одно “событие GPS” — именно поездка целиком. Это base grain — самая детальная гранулярность, доступная в источнике.

С базового грейна можно агрегировать вверх:

• one trip -> one day’s revenue.
• one trip -> one zone’s pickup_count.
• one trip -> one driver’s earnings.

Но обратно нельзя: если уже сагрегировали в день, не вернёшь поездку обратно. Поэтому храним base grain в DWH, агрегируем в marts.

2. Dimensional model: facts и dimensions

NYC Taxi — классический dimensional case. Распишем по Kimball-методике:

Fact таблица: fct_trips

Что: одна строка = одна поездка.

Колонки:

• Foreign keys (FK) к dimensions: pickup_date_key, pickup_location_key, dropoff_location_key, vendor_key, payment_type_key.
• Measures (numeric metrics): fare_amount, tip_amount, total_amount, trip_distance, trip_duration_seconds.
• Degenerate dimensions: trip_id.

Dimension таблицы:

• dim_date — конкально, день/неделя/месяц/квартал/год, праздничный ли.
• dim_location — taxi zones (LocationID, borough, zone name, service zone).
• dim_vendor — Verifone (1) / Curb (2) — taxi vendor.
• dim_payment_type — credit card / cash / no charge / dispute / unknown.

Это classic star schema — все dimensions вокруг одной fact. Снежинка (snowflake) — когда дименсии сами имеют связи между собой — здесь не нужна.

Slowly changing dimensions

В NYC Taxi dimensions относительно стабильны:

• Vendor — почти не меняется.
• Location zones — могут переименоваться, но редко.
• Payment types — фиксированный список.

Поэтому SCD Type 1 (overwrite on change) достаточно. Если бы менялись часто и нужна была история — Type 2 (с from_date / to_date / is_current).

3. Партиционирование

Партиционирование — фундаментальная оптимизация для аналитических запросов.

Что партиционируем: fct_trips — это growing-таблица, миллионы строк за каждый месяц. Партиции — must.

По какому полю: pickup_date (или его year/month). Большинство аналитических запросов фильтруют по дате (“сколько trips в мае”, “топ zones за последний месяц”). Партиционирование по date позволяет partition pruning — читать только нужные партиции.

Гранулярность: для NYC Taxi оптимально по месяцу (year_month партиции). Один файл в партиции — 1-3 GB, удобно. Партиционирование по дню даст 30x больше мелких файлов — потеря performance. По году — слишком крупное, не даёт pruning.

s3://lake/fct_trips/year_month=2024-01/data.parquet  (3 GB)
s3://lake/fct_trips/year_month=2024-02/data.parquet  (2.8 GB)
s3://lake/fct_trips/year_month=2024-03/data.parquet  (3.1 GB)
...

В DuckDB или Postgres партиционирование делается через PARTITION BY clause при создании таблицы. В data lake (S3) — через partitioned-write в Spark / pandas + pyarrow.

4. Идемпотентность

Каждый task в pipeline должен быть idempotent — иначе backfill или retry разрушат данные.

Ingester

Задача: скачать NYC Taxi Parquet за месяц YYYY-MM и положить в DWH или S3.

Идемпотентный паттерн:

def ingest_month(year_month: str):
    # 1. Detereminstic source URL
    url = f"https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{year_month}.parquet"

    # 2. Скачать в локальный tmp file
    response = requests.get(url, stream=True)
    with open(f"/tmp/trips_{year_month}.parquet", "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)

    # 3. Прочитать через pandas / pyarrow
    df = pd.read_parquet(f"/tmp/trips_{year_month}.parquet")

    # 4. Записать в staging-таблицу: DELETE по месяцу + INSERT
    conn.execute(f"DELETE FROM stg_trips WHERE year_month = '{year_month}'")
    df["year_month"] = year_month
    df.to_sql("stg_trips", conn, if_exists="append", index=False)

Что обеспечивает идемпотентность:

• URL детерминирован от year_month.
• DELETE WHERE year_month = X + INSERT — перезаписывает партицию за конкретный месяц, не затрагивая остальные.
• При retry за тот же месяц — те же данные.

Transformations (dbt)

dbt materialization для fct_trips будет incremental со strategy delete+insert по year_month:

{{ config(
    materialized='incremental',
    incremental_strategy='delete+insert',
    unique_key='year_month',
    partition_by={'field': 'year_month'}
) }}

SELECT * FROM {{ ref('stg_trips') }}

{% if is_incremental() %}
  WHERE year_month IN ({{ var('months_to_load') }})
{% endif %}

Этот pattern даёт: при первом запуске — full load всех месяцев; при последующих — только указанные месяцы перезаписываются.

5. Тесты

Тесты — встроены в dbt через tests: блоки в schema.yml.

Критичные тесты:

models:
  - name: fct_trips
    columns:
      - name: trip_id
        tests:
          - unique
          - not_null
      - name: pickup_datetime
        tests:
          - not_null
      - name: total_amount
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= 0"
      - name: pickup_location_key
        tests:
          - relationships:
              to: ref('dim_location')
              field: location_key

Что покрывают:

• unique trip_id — нет дублей.
• not_null pickup_datetime — обязательное поле.
• total_amount >= 0 — бизнес-инвариант (не может быть отрицательной выручки).
• relationships — FK integrity: каждое pickup_location_key должно существовать в dim_location.

Singular tests (custom SQL) — для более сложных проверок:

-- tests/test_no_future_trips.sql
SELECT * FROM {{ ref('fct_trips') }}
WHERE pickup_datetime > CURRENT_DATE

Если запрос возвращает строки — тест fail. Это проверяет инвариант “нет поездок в будущем”.

6. Observability

Что мониторить:

• Freshness: MAX(pickup_datetime) vs ожидаемое.
• Volume: COUNT(*) per month vs moving average.
• Null rate: % null в required-полях.
• Schema drift: список колонок vs expected.

В простом случае — это SQL-запросы, которые запускаются после каждого dbt run и пишут в metrics.pipeline_health таблицу:

INSERT INTO metrics.pipeline_health
SELECT
  'fct_trips' AS table_name,
  CURRENT_TIMESTAMP AS measured_at,
  MAX(pickup_datetime) AS max_trip_time,
  COUNT(*) AS row_count,
  SUM(CASE WHEN pickup_datetime IS NULL THEN 1 ELSE 0 END) * 1.0 / COUNT(*) AS null_rate_pickup
FROM fct_trips
WHERE year_month = '{{ var('current_month') }}';

Алерт — простейший: если max_trip_time старше 36 часов, отправить Slack-сообщение.

Диаграмма архитектуры

Чек-лист дизайна

Перед началом кода у тебя должно быть:

• Грейн фактовой таблицы определён в одном предложении.
• Список dimensions с ключевыми колонками.
• Поле для партиционирования и обоснование.
• Стратегия идемпотентности на каждый task.
• Список ключевых тестов (минимум 5).
• Список метрик observability (минимум 3).
• Диаграмма архитектуры (хотя бы на бумаге).

Если на любой пункт ответ “не знаю” — потрать ещё 30 минут на размышления. Это сэкономит часы переделок в коде.

Попробуй сам

1. Возьми pet-project на другой источник (GitHub Events / Crypto / Weather). Распиши свою dim-fact модель. Какой грейн? Какие dimensions?
2. Нарисуй диаграмму архитектуры на бумаге или в Excalidraw. Покажи кому-то и попроси задать 3 вопроса. Чему-то это научит.
3. Сравни Type 1 vs Type 2 SCD для dim_location (NYC zones периодически переименовываются). Когда нужно сохранять историю?