mini-lakehouse-lab/notebooks/05_silver_layer.ipynb

Модуль 5. Слой silver и воспроизводимые трансформации

В этом модуле мы переходим от bronze-таблицы «как есть» к осознанно очищенной и обогащённой silver-таблице.

Цели модуля:

• понять, зачем нужен silver-слой и чем он отличается от bronze;
• научиться формулировать правила трансформации явно перед написанием кода;
• отфильтровать некачественные записи, нормализовать типы и обработать NULL;
• обогатить данные через JOIN с lookup-таблицей;
• выполнить проверки качества результата на трёх уровнях: строки, схема, агрегаты.

Prerequisite:

• пройден Модуль 4: таблицы lakehouse.bronze.nyc_taxi_yellow и lakehouse.bronze.taxi_zone_lookup (самостоятельное задание) существуют.

Связка с привычным DWH (PostgreSQL / Greenplum)

	Классический DWH (Greenplum)	Lakehouse
Источник	ods/stg-таблица	Bronze Iceberg-таблица
Результат	dds/fact-таблица с очищенными данными	Silver Iceberg-таблица
Правила трансформаций	Mapping specification / ETL-документация	Явный реестр правил в ноутбуке перед кодом
Проверки качества	dq-check скрипты, data quality framework	Asserts на трёх уровнях
Воспроизводимость	Повторный запуск ETL	CREATE OR REPLACE TABLE ... AS SELECT

Ключевая идея: silver = bronze + явные правила трансформации + проверки качества.

1. Spark-сессия и проверка bronze

Инициализируем Spark и убеждаемся, что обе bronze-таблицы на месте. Теперь мы работаем только с ними и не обращаемся к raw-файлам напрямую.

In [ ]:

import pandas as pd
from pyspark.sql import SparkSession, functions as F

CATALOG_NAME = "lakehouse"
BRONZE_NAMESPACE = f"{CATALOG_NAME}.bronze"
BRONZE_TABLE = f"{BRONZE_NAMESPACE}.nyc_taxi_yellow"
BRONZE_LOOKUP_TABLE = f"{BRONZE_NAMESPACE}.taxi_zone_lookup"

SILVER_NAMESPACE = f"{CATALOG_NAME}.silver"
SILVER_TABLE = f"{SILVER_NAMESPACE}.nyc_taxi_yellow"

spark = SparkSession.builder \
    .appName("module-05-silver-layer") \
    .getOrCreate()

spark.sparkContext.setLogLevel("ERROR")

print("Spark готов")
print(f"Bronze table: {BRONZE_TABLE}")
print(f"Silver table: {SILVER_TABLE}")

In [ ]:

assert spark.catalog.tableExists(BRONZE_TABLE), (
    f"Таблица {BRONZE_TABLE} не найдена. Вернись в Модуль 4 и выполни загрузку в bronze."
)

assert spark.catalog.tableExists(BRONZE_LOOKUP_TABLE), (
    f"Таблица {BRONZE_LOOKUP_TABLE} не найдена. "
    "Вернись в Модуль 4 и выполни Секцию 8."
)

bronze_count = spark.table(BRONZE_TABLE).count()
assert bronze_count > 0, f"Таблица {BRONZE_TABLE} пуста."

print(f"Обе bronze-таблицы на месте. В основной таблице {bronze_count:,} строк. Теперь строим silver.")

2. Профиль bronze — что нужно трансформировать

Прежде чем писать трансформации, понимаем текущее состояние данных. В Модуле 3 мы уже нашли аномалии. Теперь смотрим через bronze.

In [ ]:

bronze_df = spark.table(BRONZE_TABLE)

profile_df = bronze_df.select(
    F.count("*").alias("total_rows"),
    (F.sum(F.when(F.col("fare_amount") < 0, 1).otherwise(0)) / F.count("*") * 100).alias("negative_fare_pct"),
    (F.sum(F.when(F.col("total_amount") > 1000, 1).otherwise(0)) / F.count("*") * 100).alias("extreme_total_pct"),
    (F.sum(F.when((F.col("tpep_pickup_datetime") < "2024-01-01") | (F.col("tpep_pickup_datetime") >= "2025-01-01"), 1).otherwise(0)) / F.count("*") * 100).alias("out_of_range_date_pct"),
    (F.sum(F.when(F.col("passenger_count").isNull(), 1).otherwise(0)) / F.count("*") * 100).alias("null_passengers_pct"),
    (F.sum(F.when(F.col("RatecodeID").isNull(), 1).otherwise(0)) / F.count("*") * 100).alias("null_ratecode_pct"),
    (F.sum(F.when(F.col("congestion_surcharge").isNull(), 1).otherwise(0)) / F.count("*") * 100).alias("null_congestion_pct"),
    (F.sum(F.when(F.col("Airport_fee").isNull(), 1).otherwise(0)) / F.count("*") * 100).alias("null_airport_fee_pct")
)

profile_pd = profile_df.toPandas().T
profile_pd.columns = ["value"]
profile_pd

Итоги профилирования:

Проблема	Описание	Что будем делать
Отрицательные тарифы	Ошибки в данных (fare_amount < 0)	Фильтровать (F1)
Экстремальные суммы	Аномалии (total_amount > 1000)	Фильтровать (F2)
Даты вне диапазона	Записи за пределами 2024 года	Фильтровать (F3)
NULL в passenger_count	Не заполненные данные	Заполнять 0 и приводить к INT (N1)
NULL в RatecodeID	Не заполненные данные	Заполнять 99 (unknown) и приводить к INT (N2)
NULL в сборах	Не заполненные надбавки (surcharges/fees)	Заполнять 0.0 (N3, N4)

3. Реестр правил трансформации

Центральная идея модуля: фиксируем все правила явно перед кодом. Это аналог mapping specification в DWH. Если через месяц спросят «почему в silver нет строк с отрицательным fare?», ответ будет в реестре, а не зарыт в коде.

#	Колонка/область	Правило	Тип	Обоснование
F1	fare_amount	fare_amount >= 0	Фильтр	Отрицательные тарифы — явная аномалия
F2	total_amount	total_amount <= 1000	Фильтр	Экстремальные суммы — выбросы, искажающие среднее
F3	tpep_pickup_datetime	[2024-01-01, 2025-01-01)	Фильтр	Учебный диапазон данных — 2024 год
N1	passenger_count	coalesce(cast(..., int), 0)	Норм.	Приводим к целому, заменяем NULL на 0
N2	RatecodeID	coalesce(cast(..., int), 99)	Норм.	Приводим к целому, заменяем NULL на 99 (Unknown)
N3	congestion_surcharge	coalesce(..., 0.0)	Норм.	Убираем NULL для корректности агрегатов
N4	Airport_fee	coalesce(..., 0.0)	Норм.	Убираем NULL для корректности агрегатов
E1	trip_duration_minutes	(dropoff - pickup) / 60	Обог.	Полезная вычисляемая метрика для анализа
E2	pickup_zone/borough	LEFT JOIN zone_lookup	Обог.	Понимаем, откуда поехала машина не по ID, а по имени
E3	dropoff_zone/borough	LEFT JOIN zone_lookup	Обог.	Понимаем, куда поехала машина не по ID, а по имени

4. Применение правил — фильтрация

Начинаем с фильтрации. Порядок осознанный: сначала фильтруем, потом нормализуем и обогащаем. Мы не тратим ресурсы на трансформацию строк, которые всё равно будут удалены.

In [ ]:

bronze_df = spark.table(BRONZE_TABLE)

filtered_df = (
    bronze_df
    .filter(F.col("fare_amount") >= 0)           # F1 
    .filter(F.col("total_amount") <= 1000)         # F2 
    .filter((F.col("tpep_pickup_datetime") >= "2024-01-01") & (F.col("tpep_pickup_datetime") < "2025-01-01")) # F3
)

filtered_count = filtered_df.count()
removed_count = bronze_count - filtered_count
removed_pct = (removed_count / bronze_count) * 100

print(f"Строк до фильтрации:    {bronze_count:,}")
print(f"Строк после фильтрации: {filtered_count:,}")
print(f"Удалено аномалий:       {removed_count:,} ({removed_pct:.2f}%)")

Обычно фильтры качества удаляют небольшую долю строк. Если фильтр убирает заметную часть данных — это сигнал перепроверить правило или источник.

5. Применение правил — нормализация

Приводим типы и обрабатываем NULL. В PostgreSQL мы бы использовали ALTER COLUMN TYPE integer, но в Lakehouse мы создаём новую версию данных между слоями.

In [ ]:

normalized_df = (
    filtered_df
    .withColumn("passenger_count", F.coalesce(F.col("passenger_count").cast("int"), F.lit(0)))        # N1 
    .withColumn("RatecodeID", F.coalesce(F.col("RatecodeID").cast("int"), F.lit(99)))               # N2 
    .withColumn("congestion_surcharge", F.coalesce(F.col("congestion_surcharge"), F.lit(0.0)))      # N3 
    .withColumn("Airport_fee", F.coalesce(F.col("Airport_fee"), F.lit(0.0)))                       # N4
)

normalized_df.select("passenger_count", "RatecodeID", "congestion_surcharge", "Airport_fee").show(5)

6. Применение правил — обогащение

Добавляем вычисляемые колонки и данные из lookup. Используем LEFT JOIN, чтобы не терять строки, если для какого-то ID не нашлось имени зоны. Это аналог денормализации в классическом DWH.

In [ ]:

zone_lookup_df = spark.table(BRONZE_LOOKUP_TABLE)

enriched_df = (
    normalized_df
    .withColumn("trip_duration_minutes", (F.unix_timestamp("tpep_dropoff_datetime") - F.unix_timestamp("tpep_pickup_datetime")) / 60) # E1 
    .join(zone_lookup_df.alias("pu_zone"), F.col("PULocationID") == F.col("pu_zone.LocationID"), "left") # E2 
    .withColumnRenamed("Borough", "pickup_borough") 
    .withColumnRenamed("Zone", "pickup_zone") 
    .drop("LocationID", "service_zone") 
    .join(zone_lookup_df.alias("do_zone"), F.col("DOLocationID") == F.col("do_zone.LocationID"), "left") # E3 
    .withColumnRenamed("Borough", "dropoff_borough") 
    .withColumnRenamed("Zone", "dropoff_zone") 
    .drop("LocationID", "service_zone")
)

print(f"Количество колонок после обогащения: {len(enriched_df.columns)}")
enriched_df.select("tpep_pickup_datetime", "trip_duration_minutes", "pickup_zone", "dropoff_zone").show(5, truncate=False)

Обрати внимание:

• Отрицательная длительность: если tpep_dropoff_datetime раньше tpep_pickup_datetime, колонка trip_duration_minutes будет отрицательной. Мы оставим это как наблюдение — это еще одна аномалия данных.
• Смешанный регистр колонок: оригинальные колонки (VendorID, PULocationID) сохраняют mixed case, тогда как наши новые колонки (pickup_zone, trip_duration_minutes) написаны в lowercase. Iceberg корректно обрабатывает такой набор колонок, но в реальных проектах лучше придерживаться единого стандарта.

7. Создание silver-таблицы

Используем CREATE OR REPLACE TABLE ... AS SELECT (CTAS) в namespace lakehouse.silver.

Важное предупреждение: CREATE OR REPLACE — это полная перезапись таблицы. Все предыдущие данные и история snapshot-ов будут потеряны. В учебном silver это осознанный компромисс ради идемпотентности: повторный запуск ноутбука пересоздаёт таблицу с чистого листа. В Модуле 7 мы увидим, как делать изменения безопаснее.

На 3 месяцах данных (~7 млн строк) эта операция может занять 2-4 минуты.

In [ ]:

spark.sql(f"CREATE NAMESPACE IF NOT EXISTS {SILVER_NAMESPACE}")

enriched_df.createOrReplaceTempView("silver_ready")

ctas_sql = f"""
CREATE OR REPLACE TABLE {SILVER_TABLE}
USING iceberg
AS
SELECT
    VendorID,
    tpep_pickup_datetime,
    tpep_dropoff_datetime,
    passenger_count,
    trip_distance,
    RatecodeID,
    store_and_fwd_flag,
    PULocationID,
    DOLocationID,
    payment_type,
    fare_amount,
    extra,
    mta_tax,
    tip_amount,
    tolls_amount,
    improvement_surcharge,
    total_amount,
    congestion_surcharge,
    Airport_fee,
    trip_duration_minutes,
    pickup_borough,
    pickup_zone,
    dropoff_borough,
    dropoff_zone
FROM silver_ready
"""

print(f"Выполняем CTAS для {SILVER_TABLE}...")
spark.sql(ctas_sql)

silver_df = spark.table(SILVER_TABLE)
print(f"Silver-таблица готова. Строк: {silver_df.count():,}")
silver_df.printSchema()

8. Проверки качества результата

Ни одна трансформация не считается завершённой без проверок.

8a: Строковый уровень (Row-level)

Проверяем, что правила фильтрации и нормализации сработали для каждой строки.

In [ ]:

assert silver_df.filter(F.col("fare_amount") < 0).count() == 0, "Найдены отрицательные тарифы!"
assert silver_df.filter(F.col("total_amount") > 1000).count() == 0, "Найдены экстремальные суммы!"
assert silver_df.filter(F.col("passenger_count").isNull()).count() == 0, "Найдены NULL в passenger_count!"
assert silver_df.filter(F.col("RatecodeID").isNull()).count() == 0, "Найдены NULL в RatecodeID!"

print("Row-level checks: OK")

8b: Схемный уровень (Schema-level)

Проверяем состав колонок и типы данных.

In [ ]:

expected_new_columns = ["trip_duration_minutes", "pickup_borough", "pickup_zone", "dropoff_borough", "dropoff_zone"]
for col in expected_new_columns:
    assert col in silver_df.columns, f"Колонка {col} отсутствует в silver!"

types = dict(silver_df.dtypes)
assert types["passenger_count"] == "int", f"Тип passenger_count должен быть int, а не {types['passenger_count']}"
assert types["RatecodeID"] == "int", f"Тип RatecodeID должен быть int, а не {types['RatecodeID']}"

print("Schema-level checks: OK")

8c: Агрегатный уровень (Aggregate-level)

Сравниваем общие показатели bronze и silver, чтобы оценить масштаб изменений.

In [ ]:

comparison = spark.sql(f"""
    WITH stats AS (
        SELECT 
            'bronze' as layer, 
            count(*) as rows, 
            avg(fare_amount) as avg_fare, 
            avg(trip_distance) as avg_dist
        FROM {BRONZE_TABLE}
        UNION ALL
        SELECT 
            'silver' as layer, 
            count(*) as rows, 
            avg(fare_amount) as avg_fare, 
            avg(trip_distance) as avg_dist
        FROM {SILVER_TABLE}
    )
    SELECT 
        *,
        CASE 
            WHEN layer = 'silver' THEN 
                (1 - rows * 1.0 / (SELECT rows FROM stats WHERE layer = 'bronze')) * 100 
            ELSE 0 
        END as filtered_pct
    FROM stats
""").toPandas()

comparison

Ориентир: если фильтры удалили больше 5% строк, это повод пересмотреть правила или проверить данные.

9. bronze vs silver — разделение ответственности

Мы прошли путь от raw-файлов к очищенной silver-таблице. Почему бы не делать всё сразу при загрузке из raw?

1. Bronze хранит оригинал. Если завтра выяснится, что правило фильтрации F1 было слишком жёстким, мы всегда можем пересоздать silver из bronze, не перечитывая исходные файлы из raw-зоны.
2. Разделение сложности. В bronze мы решаем задачу «прочитать и сохранить». В silver мы решаем задачу «очистить и обогатить». Это делает пайплайны проще в отладке.
3. Единый источник правды. Все downstream-потребители (аналитики, ML-инженеры) должны идти в silver. Bronze — это «внутренняя кухня» дата-инженера.

Свойство	Bronze	Silver
Состав данных	«Как в источнике»	Очищенные и обогащённые
NULL-значения	Сохранены	Обработаны (coalesce)
Типы данных	Технические (из raw)	Бизнес-ориентированные (INT для счетчиков)
Доп. колонки	Нет	Вычисляемые (длительность) и lookup (зоны)
Проверки качества	Минимальные (count)	Всесторонние (row/schema/agg)
Можно пересоздать из	Raw-файлов	Bronze-таблицы

10. Самостоятельное задание

Расширь silver-пайплайн тремя новыми правилами, пересоздай таблицу и проверь результат.

Фильтрация — F4: удалить строки с trip_distance = 0 AND total_amount != 0 (аномалия, найденная в Модуле 3).

Нормализация — N5: привести payment_type из BIGINT в INT и заменить NULL на 0 (Unknown).

Обогащение — E4: добавить вычисляемую колонку speed_mph — средняя скорость поездки в милях в час: trip_distance / (trip_duration_minutes / 60). Подсказка: обработай деление на ноль — если trip_duration_minutes = 0, результат должен быть NULL, а не ошибка.

Проверки: после пересоздания silver напиши row-level assert для F4 и schema-level проверку для speed_mph и payment_type.

In [ ]:

# Ваш код: фильтрация F4

In [ ]:

# Ваш код: нормализация N5

In [ ]:

# Ваш код: обогащение E4

In [ ]:

# Ваш код: пересоздание silver-таблицы (CTAS)

In [ ]:

# Ваш код: новые проверки качества

In [ ]:

# Ваш код: сравнение количества строк до и после добавления F4

11. Checkpoint

Проверь себя:

1. Чем silver отличается от bronze?
2. Зачем фиксировать правила перед кодом?
3. Какие три уровня проверок качества и зачем они нужны?
4. Почему мы использовали LEFT JOIN, а не INNER JOIN с lookup-таблицей?
5. Что делать, если фильтр внезапно удалил 50% строк?
6. Можно ли пересоздать silver, если через месяц мы обнаружим новую аномалию в данных?

12. Завершение

Мы не удаляем silver-таблицу. В Модуле 6 мы будем читать её одновременно через Spark и Trino, а в Модуле 8 она понадобится для финальной практики.

In [ ]:

spark.stop()