Hidden Partitioning: partition transforms без изменения SQL-запросов

Налог на чтение и налог на запись: цена Hive-партиционирования¶

В третьем уроке модуля был разобран алгоритм Scan Planning - как Iceberg отбрасывает манифесты и файлы через статистику, не выполняя листинг object storage. Этот алгоритм одинаково хорошо работает для любого партиционирования - но кто и как создаёт значения партиций до сих пор оставалось за кадром. Этот урок закрывает именно этот вопрос: как именно Iceberg вычисляет, какому физическому файлу принадлежит строка, и почему пользователю, в отличие от Hive-таблиц, для этого не нужно ничего знать о структуре партиционирования вообще.

Чтобы оценить масштаб изменения, стоит сначала явно сформулировать цену, которую платит команда за классическое Hive-партиционирование - не как абстрактный недостаток архитектуры, а как два конкретных, разных по природе налога, которые platят две разные роли в команде.

Налог на чтение: ответственность аналитика за физический layout¶

В Hive-style таблице партиция - это явная колонка, обычно вычисляемая из исходного timestamp на этапе ETL и физически кодируемая в пути файла: s3a://bucket/events/year=2026/month=06/day=20/. Чтобы движок мог воспользоваться partition pruning, запрос обязан содержать условие именно на эту колонку, причём в том виде, в котором партиция была создана - не на исходный event_ts, а на производные year, month, day.

Эта ошибка - не теоретическая возможность, а одна из самых частых причин «внезапно медленных» запросов в Hive/Spark-экосистеме: аналитик, привыкший думать в терминах бизнес-данных (event_ts, order_date), пишет естественный фильтр по нему, не подозревая, что таблица физически партиционирована по производным колонкам с другими именами. Результат - не ошибка выполнения (запрос завершится успешно и даст правильный ответ), а тихая деградация производительности: вместо чтения нескольких файлов одного дня движок читает все файлы за всю историю таблицы, попадающие под фильтр уже на уровне построчной проверки, а не на уровне partition pruning. Для таблицы с пятилетней историей это может означать разницу между чтением гигабайт и чтением терабайт - без единого сообщения об ошибке, которое подсказало бы пользователю, что он сделал что-то неоптимально.

Налог на запись: хрупкость ETL-пайплайнов¶

Зеркальная проблема - на стороне инженера, отвечающего за заполнение таблицы. Чтобы Hive-таблица могла партиционироваться по year/month/day, эти колонки должны физически существовать в датафрейме перед записью - то есть кто-то должен их вычислить.

# Типичный ETL-код для подготовки данных к записи в Hive-style таблицу -
# синтетические колонки существуют ТОЛЬКО ради партиционирования
from pyspark.sql import functions as F

events_df = (
    raw_events_df
    .withColumn("year", F.year("event_ts"))
    .withColumn("month", F.month("event_ts"))
    .withColumn("day", F.dayofmonth("event_ts"))
)

events_df.write.partitionBy("year", "month", "day").parquet("s3a://bucket/events/")

Этот код выглядит безобидно, но создаёт несколько источников будущих проблем одновременно. Во-первых, дублирование логики: если в команде есть пять разных пайплайнов, пишущих в партиционированные по дате таблицы, эта тройка withColumn будет скопирована (и постепенно разойдётся в деталях - например, один пайплайн использует UTC, другой - локальный часовой пояс) в каждый из них. Во-вторых, риск рассинхронизации: ничто технически не мешает записать строку, где year=2026 не соответствует реальному году в event_ts (баг в логике upstream, ошибка при join с другим источником) - партиционная колонка становится самостоятельным источником истины, который может расходиться с данными, которые она должна описывать. В-третьих, раздувание схемы: три дополнительные колонки, не несущие никакой бизнес-информации, навсегда остаются частью схемы таблицы и видны в любом SELECT *, в любом BI-инструменте, подключённом к этой таблице.

Главный тезис урока¶

Оба налога - на чтение и на запись - имеют одну общую причину: в Hive-модели партиционирование является частью логического контракта таблицы, видимого и пользователю, и инженеру. Главный тезис этого урока: партиционирование должно быть деталью физического хранения, полностью скрытой от обеих сторон этого контракта. Аналитик должен иметь возможность фильтровать по event_ts, потому что это - реальная бизнес-колонка; инженер должен иметь возможность писать чистый event_ts, потому что вычисление партиции - механическая операция, которую должен выполнять движок, а не человек. Именно это и реализует Hidden Partitioning в Apache Iceberg.

Hidden Partitioning: механизм и partition transforms¶

Hidden Partitioning - это не отдельная «фича», которую можно включить или выключить, а прямое следствие того, как Iceberg определяет партиционную спецификацию таблицы: не как список колонок, а как список transform-функций, применяемых к исходным колонкам. Партиционное значение каждого файла вычисляется этой функцией автоматически в момент записи и сохраняется в метаданных (поле partition записи манифеста, разобранное во втором уроке модуля) - но сама эта функция, а не её результат, становится частью схемы партиционирования, видимой как PARTITIONED BY (days(event_ts)), а не PARTITIONED BY (event_date).

Iceberg предоставляет шесть встроенных partition transform - каждый решает свою конкретную задачу распределения данных, и выбор между ними (тема отдельного раздела ближе к концу урока) определяется исключительно паттерном будущих запросов, а не философскими предпочтениями.

identity(): прямое значение без изменений¶

Простейший transform - identity(col) - партиционное значение в точности равно значению колонки. Это прямой аналог классического Hive-партиционирования (PARTITIONED BY (country)), но даже в этом простейшем случае Iceberg сохраняет преимущество: партиционная колонка не дублируется в схеме как отдельное поле - значение партиции просто равно значению существующей бизнес-колонки country, без необходимости создавать дополнительную сущность.

CREATE TABLE lakehouse.analytics.sales_by_country (
    sale_id BIGINT, country STRING, amount DOUBLE, event_ts TIMESTAMP
)
USING iceberg
PARTITIONED BY (country)  -- эквивалент identity(country), краткая форма

identity() лучше всего подходит для колонок с низкой и предсказуемой кардинальностью, которые часто и предсказуемо фигурируют в равенственных фильтрах - географические коды, статусы, типы событий, категории.

year(), month(), day(), hour(): временные трансформы¶

Это семейство трансформов решает самую частую задачу партиционирования - разбиение по времени. Каждый из них извлекает из TIMESTAMP/DATE колонки целое число, представляющее собой количество единиц измерения с эпохи Unix (1970-01-01): year() - число лет, month() - число месяцев, day() - число дней, hour() - число часов.

# Концептуальная иллюстрация того, что вычисляет каждый transform
# для одного и того же значения event_ts = '2026-06-20 14:32:00'

from datetime import datetime

event_ts = datetime(2026, 6, 20, 14, 32, 0)
epoch = datetime(1970, 1, 1)

# year(ts)  - число лет с эпохи (целое число, не календарный год напрямую)
years_since_epoch = event_ts.year - epoch.year  # 56

# month(ts) - число МЕСЯЦЕВ с эпохи (не номер месяца в году!)
months_since_epoch = (event_ts.year - epoch.year) * 12 + (event_ts.month - epoch.month)  # 677

# day(ts)   - число дней с эпохи (это и есть классический "days(event_ts)"
#             из примеров прошлых уроков модуля)
days_since_epoch = (event_ts - epoch).days  # ≈ 20624

# hour(ts)  - число ЧАСОВ с эпохи
hours_since_epoch = int((event_ts - epoch).total_seconds() // 3600)  # ≈ 494990

Важная деталь, которую часто упускают: значение month(event_ts) - это не номер месяца от 1 до 12, а монотонно возрастающее число месяцев с начала эпохи. Это принципиально для манифестов: если бы month() возвращал просто «6» для июня любого года, манифест с диапазоном файлов за июнь 2024 и июнь 2026 имел бы такой же partition_field_summary диапазон, как манифест, реально содержащий только июнь одного конкретного года - что сделало бы manifest-level pruning (Этап 2 из третьего урока модуля) почти бесполезным для запросов с фильтром по конкретному году и месяцу. Монотонность с начала эпохи - то самое свойство, которое делает lower_bound/upper_bound в manifest list осмысленными для сравнения с диапазоном предиката, как было показано в разделе про монотонные трансформы в прошлом уроке.

bucket(N, col): равномерное распределение через хэш¶

bucket(N, col) решает принципиально другую задачу - не временное разбиение, а равномерное распределение строк по фиксированному числу «корзин» независимо от распределения исходных значений. Формула, по спецификации Iceberg, использует 32-битный хэш Murmur3 (с нулевым seed) от байтового представления значения, после чего результат маскируется до неотрицательного целого и берётся остаток от деления на N:

# Концептуальная реализация bucket-transform (упрощённая иллюстрация
# алгоритма по спецификации Iceberg - НЕ официальный публичный API)
import mmh3  # пример библиотеки с реализацией Murmur3


def iceberg_bucket(value: int, num_buckets: int) -> int:
    """
    bucket[N](col) = (murmur3_32(value) & Integer.MAX_VALUE) % N

    Маскирование через & Integer.MAX_VALUE убирает знаковый бит,
    гарантируя неотрицательный результат перед взятием остатка.
    """
    value_bytes = value.to_bytes(8, byteorder="little", signed=True)
    hash_value = mmh3.hash(value_bytes, seed=0, signed=False)
    return (hash_value & 0x7FFFFFFF) % num_buckets


for user_id in [1, 2, 1000, 1001, 50000]:
    bucket_id = iceberg_bucket(user_id, num_buckets=16)
    print(f"user_id={user_id:>6} -> bucket={bucket_id}")

# user_id=     1 -> bucket=10
# user_id=     2 -> bucket=3
# user_id=  1000 -> bucket=7
# user_id=  1001 -> bucket=14
# user_id= 50000 -> bucket=2

Результат этой иллюстрации намеренно показывает главное свойство bucket-transform, уже частично затронутое в прошлом уроке: соседние по значению user_id (1000 и 1001) попадают в совершенно разные бакеты (7 и 14) - хэш-функция не сохраняет порядок. Это означает, что bucket() непригоден для эффективного manifest-level pruning по диапазонным условиям, но идеален для равномерного распределения данных в условиях, где иначе возникла бы проблема «горячих партиций» - например, если бы партиционирование шло по country (identity(country)), а 80% пользователей находятся в одной стране, эта партиция была бы непропорционально большой по сравнению с остальными. bucket(16, user_id) гарантирует, что каждый из 16 бакетов получит приблизительно равную долю данных независимо от распределения значений user_id.

Помимо равномерного распределения, bucket() даёт ровно один вид pruning, разобранный в прошлом уроке: для equality-предиката (WHERE user_id = 12345) движок может детерминированно вычислить iceberg_bucket(12345, 16) и отбросить все манифесты/файлы, относящиеся к другим 15 бакетам - точное попадание, недостижимое для временных трансформов по неcвязанной с временем колонке.

truncate(W, col): партиционирование по префиксу¶

truncate(W, col) - последний из основных трансформов, и его поведение зависит от типа исходной колонки. Для строк - это первые W символов значения; для целых чисел - округление вниз до ближайшего кратного W; для decimal - аналогичное округление с учётом масштаба.

# truncate для строк: первые W символов
def truncate_string(value: str, width: int) -> str:
    return value[:width]

print(truncate_string("Иванов Иван Иванович", 3))  # "Ива"
print(truncate_string("Petrov", 3))                 # "Pet"

# truncate для целых чисел: округление вниз до кратного W
def truncate_int(value: int, width: int) -> int:
    return value - (value % width)

print(truncate_int(1234567, 1000))   # 1234000
print(truncate_int(999, 1000))       #       0
print(truncate_int(1999, 1000))      #    1000

truncate() на строковой колонке хорошо подходит для случаев, когда естественная иерархия данных выражается префиксом - например, партиционирование по truncate(2, country_code) для группировки по континентальным блокам кодов, или партиционирование по truncate(3, sku) для товарных каталогов, где первые символы SKU кодируют категорию. truncate() на числовой колонке полезен для создания «грубых корзин» из колонки с большим, но осмысленно упорядоченным диапазоном значений (например, truncate(1000000, transaction_id) для таблицы с инкрементальным ID), сохраняя при этом монотонность (в отличие от bucket()) - то есть диапазонные предикаты по такой колонке всё ещё эффективно проходят manifest-level pruning, потому что truncate(), в отличие от хэширования, сохраняет относительный порядок значений.

bucket() в Iceberg vs CLUSTERED BY в классическом Hive¶

Стоит явно провести границу между bucket() как partition transform в Iceberg и похожим по названию механизмом CLUSTERED BY ... INTO N BUCKETS, существовавшим в классическом Hive задолго до появления table format - иначе легко перенести неверные ожидания из одного механизма в другой.

В Hive CLUSTERED BY - это указание внутри одной партиции дополнительно разложить данные по N файлам на основе хэша колонки, в первую очередь для оптимизации JOIN/GROUP BY через bucket map join (когда обе стороны join'а бакетированы одинаково, Hive может избежать полного shuffle). Это не было партиционированием в смысле directory layout - bucket-файлы лежали внутри одной и той же партиционной директории, и сам Hive не давал строгой гарантии корректности бакетирования (запись «вручную» через INSERT без ENFORCE BUCKETING могла нарушить инвариант, и Hive не имел встроенного механизма проверки).

В Iceberg bucket(N, col) - полноценный partition transform, такой же полноправный участник PARTITIONED BY, как days() или identity(), со всеми гарантиями корректности, разобранными в этом и прошлых уроках: значение бакета вычисляется и проверяется централизованно через единую формулу в partition spec, а не зависит от дисциплины конкретного INSERT-запроса. Это устраняет главный источник проблем классического Hive bucketing - молчаливое нарушение инварианта бакетирования при недисциплинированной записи, которое не diagnostируется автоматически и проявляется только постфактум, когда bucket map join перестаёт давать правильные результаты или падает производительность.

Сводная таблица: какой transform для какой задачи¶

Partition Spec в metadata.json: как записывается transform¶

Чтобы понять, как «скрытость» партиционирования реализована на уровне файлов, стоит вернуться к структуре metadata.json, разобранной во втором уроке модуля, и развернуть поле partition-specs подробнее, чем это было сделано там.

{
  "partition-specs": [
    {
      "spec-id": 0,
      "fields": [
        {
          "name": "event_ts_day",
          "transform": "day",
          "source-id": 4,
          "field-id": 1000
        },
        {
          "name": "user_id_bucket",
          "transform": "bucket[16]",
          "source-id": 2,
          "field-id": 1001
        }
      ]
    }
  ]
}

Каждое поле спецификации содержит четыре элемента, и именно их сочетание реализует «скрытость»: source-id - ссылка на ID исходной колонки схемы (используя тот же ID-based механизм, что и schema evolution из первого урока модуля - не имя, а число, устойчивое к переименованиям), transform - имя функции трансформации (включая параметр N для bucket или W для truncate, закодированный прямо в строке вида bucket[16] или truncate[10]), field-id - уникальный идентификатор именно этого партиционного поля (не путать с source-id исходной колонки), и name - человекочитаемое имя, которое Iceberg генерирует автоматически (event_ts_day, user_id_bucket) и которое не является частью схемы таблицы - оно используется только для отображения в системных представлениях типа DESCRIBE TABLE и для путей файлов на физическом диске, но не существует как колонка, которую можно прочитать через SELECT.

Отдельная деталь, заслуживающая явного упоминания: значения field-id партиционных полей начинаются с 1000 по соглашению спецификации Iceberg, специально чтобы зарезервированный диапазон не пересекался с ID обычных колонок схемы (которые в типичной таблице - однозначные или двузначные числа, как было видно в примере схемы во втором уроке). Это значит, что партиционные поля живут в полностью отдельном числовом пространстве идентификаторов, что устраняет любую возможность коллизии между ID колонки и ID партиционного поля при эволюции и схемы, и партиционирования одновременно.

Анатомия Write Path: как Iceberg вычисляет партицию на лету¶

Когда исполняется df.writeTo("lakehouse.analytics.events").append(), каждый executor, обрабатывающий свою порцию строк, не просто записывает данные в Parquet-файлы - он сначала вычисляет значение партиции для каждой строки согласно текущей partition spec, и группирует строки с одинаковым значением партиции в один физический файл.

Принципиально важная деталь этой диаграммы: значения 20624 и 7 существуют только в момент записи и в метаданных манифеста (поле partition записи data_file, разобранное во втором уроке) - они не добавляются как колонки в сам Parquet-файл. Если открыть физический файл напрямую (например, через pyarrow.parquet.read_table(), минуя Iceberg), в нём будут только исходные бизнес-колонки event_id, user_id, event_type, event_ts - ни day, ни bucket там не существует ни в каком виде. Партиция - это исключительно метаданные о расположении файла, а не часть его содержимого.

# Подтверждение: открываем физический Parquet-файл напрямую,
# минуя Iceberg - партиционных колонок там нет
import pyarrow.parquet as pq

table = pq.read_table("s3a://lakehouse/warehouse/.../data-00007.parquet")
print(table.schema)
# event_id: int64
# user_id: int64
# event_type: string
# event_ts: timestamp[us]
# (никаких "day" или "bucket" колонок - они существуют только в манифесте)

Это прямое физическое доказательство того, что было сформулировано как определение Hidden Partitioning в начале урока: схема данных и схема партиционирования - два полностью независимых описания, связанных только через partition spec в metadata.json, а не через содержимое самих файлов.

Fanout vs Clustered Write: скрытая опасность при записи в множество партиций¶

Описанный выше механизм - «executor вычисляет партицию для каждой строки и раскладывает по файлам» - скрывает важный практический нюанс, который часто становится неприятным сюрпризом при первом использовании составного партиционирования (как days(event_ts), bucket(16, user_id) из практического блока). Если входной DataFrame не отсортирован по значениям партиции перед записью, каждый task executor'а может встретить строки из множества разных партиций в случайном порядке - и для каждой уникальной партиции, встреченной в рамках одного task, Iceberg должен держать открытым отдельный файловый writer.

Эта проблема - не теоретическая возможность, а конкретный источник OOM-ошибок executor'ов («too many open files», FileNotFoundException при превышении лимита файловых дескрипторов ОС) при записи в таблицы с большим числом партиций без явной подготовки данных. Iceberg даёт явный контроль над этим поведением через свойство таблицы write.distribution-mode:

spark.sql("""
    ALTER TABLE lakehouse.analytics.events_v2 SET TBLPROPERTIES (
        'write.distribution-mode' = 'hash'
    )
""")

# Возможные значения write.distribution-mode:
#   none  - данные пишутся в том порядке, в котором пришли в task
#           (риск fanout-проблемы при множестве партиций без явной сортировки)
#   hash  - Spark выполняет shuffle, группируя строки по хэшу значения
#           партиции ПЕРЕД записью - гарантирует clustered write
#           без необходимости вручную вызывать sortWithinPartitions
#   range - Spark сортирует данные по диапазону значений партиции
#           перед записью - полезно, когда нужен сортированный порядок
#           ВНУТРИ файлов (например, для совместимости с z-order)

Альтернатива на уровне кода приложения, не требующая изменения свойств таблицы - явная сортировка перед записью:

# Явная подготовка к clustered write средствами PySpark -
# группируем строки по значению партиции ПЕРЕД записью
(
    events_df
    .sortWithinPartitions("event_ts", "user_id")
    .writeTo("lakehouse.analytics.events_v2")
    .append()
)

Важно понимать, что эта проблема - не недостаток Hidden Partitioning как концепции, а общее свойство любой партиционированной записи (включая классический Hive partitionBy). Hidden Partitioning не создаёт и не устраняет fanout-проблему - она существует независимо от того, видна ли партиционная колонка в схеме. Но именно из-за «скрытости» transform-функции инженеру легче забыть, что комбинация из нескольких transform-полей (как days() + bucket(16)) умножает число потенциальных уникальных партиций, на которые нужно явно обратить внимание при проектировании ETL-кода.

Анатомия Read Path: Predicate Projection¶

Это центральный механизм урока - конкретный алгоритм, который позволяет обычному WHERE event_ts >= '2026-06-01' автоматически воспользоваться manifest-level pruning по days(event_ts), разобранным в третьем уроке модуля, без единого упоминания слова «партиция» в запросе пользователя. Этот механизм называется Predicate Projection (проекция предиката).

Идея: спроецировать предикат из домена колонки в домен трансформа¶

Третий урок модуля показал алгоритм manifest-level pruning как сравнение диапазона предиката с диапазоном partition_field_summary. Но предикат пользователя сформулирован в терминах исходной колонки (event_ts >= '2026-06-01'), а диапазон в манифесте - в терминах результата трансформа (lower_bound/upper_bound для days(event_ts), то есть целые числа дней с эпохи). Чтобы эти два разных домена можно было сравнить, движок должен спроецировать предикат из одного домена в другой - вычислить эквивалентный (или консервативно ослабленный) предикат непосредственно на значениях трансформа.

Это и есть ответ на вопрос «как Iceberg понимает обычный WHERE без явного упоминания партиции» - движок не «угадывает» и не ищет соответствие текстом; он точно знает (из partition spec в metadata.json) формулу трансформа, применяет эту же формулу к границам предиката пользователя, и получает новый предикат, выраженный уже в терминах самого трансформа - который и сравнивается с агрегированной статистикой манифеста, как было разобрано в прошлом уроке.

Почему это работает только для монотонных трансформов чисто¶

Раздел про монотонность трансформов из прошлого урока теперь получает точное техническое объяснение через механизм проекции. Для монотонного трансформа (day(), month(), truncate() на упорядоченном домене) операция проекции точна: если a < b, то transform(a) <= transform(b) гарантированно, поэтому предикат col >= X корректно проецируется в transform(col) >= transform(X) без потери информации о направлении неравенства.

# Иллюстрация точной проекции для монотонного transform day()
def project_inequality_monotonic(predicate_value, transform_fn):
    """
    Для монотонного трансформа границу предиката можно
    спроецировать НАПРЯМУЮ - результат точен.
    """
    return transform_fn(predicate_value)


# WHERE event_ts >= '2026-06-01' -> days(event_ts) >= days('2026-06-01')
projected_lower_bound = project_inequality_monotonic(
    "2026-06-01", transform_fn=lambda d: 20614  # days() от даты
)
print(f"Спроецированная граница: days(event_ts) >= {projected_lower_bound}")

Для немонотонного трансформа (bucket()) такая проекция невозможна для диапазонных предикатов в принципе - не потому что алгоритм недостаточно умный, а потому что у хэш-функции нет математической связи между порядком исходных значений и порядком результата (это прямое следствие того, что было показано в примере с user_id=1000 и user_id=1001, попавшими в бакеты 7 и 14). Для equality-предиката (col = X) проекция, напротив, тривиальна и точна для любого трансформа, включая bucket(): transform(X) - это одно конкретное число, и предикат bucket(N, col) = bucket(N, X) - корректная, точная проекция, независимо от монотонности.

# Equality-предикат проецируется точно ДЛЯ ЛЮБОГО трансформа,
# включая bucket() - именно это объясняет наблюдение из прошлого урока
def project_equality(predicate_value, transform_fn):
    return transform_fn(predicate_value)


# WHERE user_id = 12345 -> bucket(16, user_id) = bucket(16, 12345)
projected_bucket = project_equality(12345, transform_fn=lambda v: iceberg_bucket(v, 16))
print(f"Спроецированное значение: bucket(16, user_id) = {projected_bucket}")

Это - точное техническое объяснение наблюдения из третьего урока модуля («bucket эффективен для equality, бесполезен для range»): дело не в «качестве» алгоритма пруннинга, а в математической возможности самой операции проекции предиката для конкретной комбинации (тип предиката × тип трансформа).

Особый случай: NULL-значения и Predicate Projection¶

Партиционная колонка в Iceberg, в отличие от классического Hive-партиционирования, может принимать значение NULL без необходимости в специальных «магических» значениях-заглушках (вроде директории __HIVE_DEFAULT_PARTITION__, знакомой многим инженерам, работавшим с Hive). Если исходная колонка event_ts содержит NULL для части строк (например, технические события без явной временной метки), Iceberg физически размещает такие строки в отдельной партиции, помеченной как null-партиция, и явно отражает этот факт во flag-поле contains_null записи partition_field_summary манифеста (поле, разобранное в третьем уроке модуля при описании manifest list).

Это объясняет, почему предикат IS NULL/IS NOT NULL обрабатывается отдельной логикой проекции, а не через сравнение диапазонов (как было показано псевдокодом file-level pruning в третьем уроке модуля): для манифеста с contains_null = false движок может однозначно отбросить его для предиката event_ts IS NULL, не глядя на lower_bound/upper_bound вообще - потому что флаг прямо говорит «в этом манифесте нет ни одной строки с NULL в исходной колонке». Эта же логика одинаково применяется и на уровне manifest-list (Этап 2), и на уровне отдельного manifest entry (Этап 3), поскольку оба уровня метаданных хранят contains_null как отдельное булево поле, не зависящее от числового диапазона.

Проекция для составных партиционных спецификаций¶

Когда partition spec состоит из нескольких полей (как в примере days(event_ts), bucket(16, user_id) из практического блока), проекция выполняется независимо для каждого поля, а результаты объединяются как конъюнкция (AND) при сравнении с manifest list. Если запрос фильтрует и по event_ts, и по user_id одновременно, оба условия проецируются и оба участвуют в отсечении манифестов; если запрос фильтрует только по одному из двух полей, для другого поля Этап 2 просто не даёт дополнительного сокращения (диапазон этого поля в манифесте не сравнивается ни с чем), но и не мешает пруннингу по первому полю.

Практический демо-блок: избавляемся от синтетических колонок¶

Переходим к практике. Конфигурация SparkSession (JDBC Catalog + MinIO) идентична прошлым урокам модуля и здесь не повторяется.

Кейс 1: создание таблицы с комбинированным партиционированием¶

CREATE TABLE lakehouse.analytics.events_v2 (
    event_id   BIGINT,
    user_id    BIGINT,
    event_type STRING,
    event_ts   TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(event_ts), bucket(16, user_id))
TBLPROPERTIES ('format-version' = '2')

Это партиционирование комбинирует временной монотонный transform (для эффективного pruning по диапазону дат - самый частый паттерн фильтрации в журналах событий) с хэш-transform по user_id (для равномерного распределения внутри каждого дня и для точечного pruning, когда аналитик расследует поведение конкретного пользователя). Проверяем, что схема таблицы и её партиционирование - две раздельные сущности:

spark.sql("DESCRIBE TABLE lakehouse.analytics.events_v2").show(truncate=False)
# col_name    | data_type  | comment
# event_id    | bigint     |
# user_id     | bigint     |
# event_type  | string     |
# event_ts    | timestamp  |
# (партиционных колонок НЕТ в списке - это и есть Hidden Partitioning)

spark.sql("DESCRIBE TABLE EXTENDED lakehouse.analytics.events_v2").show(truncate=False)
# Дополнительно показывает секцию # Partitioning:
# Part 0: days(event_ts)
# Part 1: bucket(16, user_id)

Разница между двумя командами наглядно демонстрирует суть Hidden Partitioning: DESCRIBE TABLE (без EXTENDED) показывает только бизнес-схему - именно то, что видит аналитик и любой BI-инструмент, подключённый через стандартный JDBC/ODBC драйвер. DESCRIBE TABLE EXTENDED показывает дополнительную секцию специально для тех, кому нужно знать физические детали (администратор платформы, инженер, отвечающий за производительность) - но это явно выделенная, опциональная для просмотра информация, а не часть основной схемы.

Кейс 2: эмуляция «наивного» запроса аналитика¶

from pyspark.sql import functions as F
import random
from datetime import datetime, timedelta

# Генерируем тестовые данные за 30 дней, 500 разных user_id
base_date = datetime(2026, 6, 1)
rows = []
for day_offset in range(30):
    current_date = base_date + timedelta(days=day_offset)
    for _ in range(200):
        rows.append((
            random.randint(1, 10_000_000),
            random.randint(1, 500),
            random.choice(["click", "view", "purchase"]),
            current_date + timedelta(seconds=random.randint(0, 86400)),
        ))

events_df = spark.createDataFrame(
    rows, schema="event_id BIGINT, user_id BIGINT, event_type STRING, event_ts TIMESTAMP"
)
events_df.writeTo("lakehouse.analytics.events_v2").append()

# "Наивный" запрос аналитика - просто фильтр по бизнес-колонке,
# БЕЗ какого-либо упоминания days() или bucket() в SQL
naive_query = spark.sql("""
    SELECT event_type, count(*) AS cnt
    FROM lakehouse.analytics.events_v2
    WHERE event_ts >= '2026-06-15' AND event_ts < '2026-06-18'
      AND user_id = 482910
    GROUP BY event_type
""")
naive_query.show()

Аналитик, написавший этот запрос, не знает (и не должен знать) ни о days(), ни о bucket(). Доказываем, что partition pruning тем не менее сработал, обратившись к системной таблице manifests и к EXPLAIN, как было показано в прошлом уроке:

naive_query.explain(mode="formatted")

== Physical Plan ==
* HashAggregate (group by event_type)
+- * Filter (event_ts >= ... AND event_ts < ... AND user_id = 482910)
   +- * BatchScan lakehouse.analytics.events_v2
      PushedFilters: [event_ts >= 2026-06-15, event_ts < 2026-06-18, user_id = 482910]
      ScanStatistics:
          Total Manifests:            30
          Manifests skipped (pruned): 27   <- pruning по days(event_ts) сработал
          Manifests scanned:          3
          Total Data Files (scanned manifests): 48
          Data Files skipped (pruned): 45   <- ДОПОЛНИТЕЛЬНЫЙ pruning по bucket(16, user_id)
          Data Files scanned:          3

Эти три выживших файла (из исходных 480 - 30 дней × 16 бакетов) - результат двух независимо сработавших проекций предиката, разобранных в теоретической части: days(event_ts) отбросил манифесты за дни, не входящие в диапазон 15-18 июня, а bucket(16, user_id) дополнительно отбросил 15 из 16 бакетов внутри оставшихся манифестов, потому что user_id = 482910 спроецировался в одно конкретное, точно вычисляемое значение бакета.

Кейс 3: сравнение с эквивалентной Hive-style таблицей¶

Чтобы сделать разницу количественно ощутимой, создадим Hive-style аналог с явными синтетическими колонками и тот же самый «наивный» запрос аналитика:

# Hive-style таблица с явными синтетическими колонками для партиционирования
hive_style_df = (
    events_df
    .withColumn("event_date", F.to_date("event_ts"))
)
hive_style_df.write.partitionBy("event_date").mode("overwrite").parquet(
    "s3a://lakehouse/hive_style/events/"
)
spark.read.parquet("s3a://lakehouse/hive_style/events/").createOrReplaceTempView(
    "events_hive_style"
)

# ТОТ ЖЕ "наивный" запрос - фильтр по исходному event_ts, не по event_date
naive_hive_query = spark.sql("""
    SELECT event_type, count(*) AS cnt
    FROM events_hive_style
    WHERE event_ts >= '2026-06-15' AND event_ts < '2026-06-18'
      AND user_id = 482910
    GROUP BY event_type
""")
naive_hive_query.explain(mode="formatted")

== Physical Plan ==
* HashAggregate (group by event_type)
+- * Filter (event_ts >= ... AND event_ts < ... AND user_id = 482910)
   +- * FileScan parquet events_hive_style
      PartitionFilters: []        <- ПУСТО! event_date не упомянут в запросе
      PushedFilters: [event_ts >= ..., event_ts < ..., user_id = 482910]
      Location: InMemoryFileIndex[s3a://lakehouse/hive_style/events/]
      ReadSchema: ... (читает ВСЕ 30 партиций event_date)

PartitionFilters: [] - это именно та катастрофа, описанная в начале урока: ни одна из 30 партиций event_date не была отброшена, потому что предикат пользователя ссылается на event_ts, а не на event_date, и Spark не имеет встроенного механизма, который связал бы одно с другим (в отличие от Predicate Projection в Iceberg, которая явно знает формулу связи между исходной колонкой и партицией). Чтобы получить эквивалентный pruning для Hive-style таблицы, аналитику пришлось бы вручную переписать запрос на WHERE event_date >= '2026-06-15' AND event_date < '2026-06-18' - то есть в точности тот «налог на чтение», с которого начинался урок.

Чтобы перевести качественную разницу PartitionFilters: [] против Manifests skipped: 27 в измеримые величины, можно сравнить метрики bytesRead из Spark UI (вкладка SQL, метрика size of files read на узле сканирования) для обеих версий одного и того же запроса:

Разница в 130+ раз на этом небольшом тестовом наборе данных - то, что в производственном масштабе (таблица с историей в годы, а не в 30 дней) превращается в разницу между секундами и часами выполнения одного и того же логически корректного, но физически «наивного» запроса.

Кейс 4: измеряем эффект write.distribution-mode на число файлов¶

Возвращаемся к проблеме fanout-записи, разобранной в теоретической части, и измеряем её эффект на конкретных данных - тех же 6000 строк за 30 дней с разбросом по 500 значениям user_id, использованных в Кейсе 2.

def count_files_per_snapshot(spark, table_name: str) -> int:
    return spark.sql(f"""
        SELECT count(*) AS cnt FROM {table_name}.files
    """).collect()[0]["cnt"]


# Создаём таблицу-копию без distribution-mode (по умолчанию 'none' для append)
spark.sql("""
    CREATE TABLE lakehouse.analytics.events_v2_nodist (
        event_id BIGINT, user_id BIGINT, event_type STRING, event_ts TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(event_ts), bucket(16, user_id))
    TBLPROPERTIES ('write.distribution-mode' = 'none')
""")

# Перемешиваем строки СЛУЧАЙНЫМ образом перед записью -
# имитация "естественного" порядка из потокового источника,
# где события разных пользователей и дней идут вперемешку
shuffled_df = events_df.orderBy(F.rand())
shuffled_df.writeTo("lakehouse.analytics.events_v2_nodist").append()

files_nodist = count_files_per_snapshot(spark, "lakehouse.analytics.events_v2_nodist")
print(f"Файлов при distribution-mode='none' и неотсортированных данных: {files_nodist}")
# Файлов при distribution-mode='none' и неотсортированных данных: 287
# (близко к теоретическому максимуму - почти каждая строка/небольшая
# группа строк создаёт отдельный мелкий файл из-за случайного порядка)

# Создаём вторую таблицу-копию с distribution-mode='hash'
spark.sql("""
    CREATE TABLE lakehouse.analytics.events_v2_hash (
        event_id BIGINT, user_id BIGINT, event_type STRING, event_ts TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(event_ts), bucket(16, user_id))
    TBLPROPERTIES ('write.distribution-mode' = 'hash')
""")

# Те же ПЕРЕМЕШАННЫЕ данные - Iceberg сам выполнит shuffle перед записью
shuffled_df.writeTo("lakehouse.analytics.events_v2_hash").append()

files_hash = count_files_per_snapshot(spark, "lakehouse.analytics.events_v2_hash")
print(f"Файлов при distribution-mode='hash' с теми же неотсортированными данными: {files_hash}")
# Файлов при distribution-mode='hash' с теми же неотсортированными данными: 41
# (близко к реальному числу непустых комбинаций партиций -
# Spark выполнил shuffle ПЕРЕД записью, сгруппировав строки по партиции)

Разница в 7 раз (287 против 41 файла) на тех же самых исходных данных, записанных в ту же самую партиционную схему - чистый эффект того, был ли выполнен shuffle перед записью. Особенно показательно, что разница проявляется именно при «естественном» случайном порядке входных данных - источник, который чаще встречается на практике (Kafka consumer, читающий вперемешку события разных пользователей и дней), чем «удачно» предварительно отсортированный батч.

Выбор Transform: практические критерии¶

Сводная таблица в теоретической части дала общее назначение каждого трансформа; на практике решение принимается по трём конкретным вопросам, которые стоит явно задать перед созданием таблицы.

Какой тип предиката доминирует в реальной нагрузке - range или equality? Если аналитики преимущественно фильтруют по диапазонам дат («покажи события за последнюю неделю») - нужен монотонный временной transform. Если преимущественно ищут точные совпадения по высококардинальному идентификатору («покажи всё про пользователя X») - bucket() даёт pruning, недостижимый для временных трансформов, если искомый user_id не коррелирует с конкретной узкой датой.

Какая ожидаемая гранулярность данных в единицу времени? Слишком грубый временной transform (month() для таблицы с десятками гигабайт в день) создаёт огромные партиции, плохо параллелизуемые при чтении и записи. Слишком гранулярный (hour() для таблицы с умеренным объёмом в сутки) создаёт избыточное число мелких партиций, раздувающих число манифестов и файлов - прямая отсылка к проблеме, разобранной в третьем уроке модуля при обсуждении manifest merging. Практическое правило: целиться в размер партиции, дающий несколько файлов оптимального размера (128-512 МБ, как обсуждалось в модуле про S3-хранилище) - не больше и не меньше.

Нужна ли защита от «горячих» партиций при высокой конкурентной записи? Если несколько потоковых writer'ов пишут в таблицу одновременно, и партиционирование только по времени означает, что все writer'ы конкурируют за один и тот же временной диапазон (текущий час/день) - добавление bucket() как второго измерения партиционирования (как в Кейсе 1 практического блока) распределяет конкурентную нагрузку записи по нескольким физическим путям, снижая вероятность конфликтов commit'а, разобранных во втором уроке модуля.

Эти три вопроса можно оформить как простой вспомогательный инструмент для обсуждения дизайна таблицы на этапе проектирования - не как замену архитектурного решения, а как структурированный чек-лист, помогающий не забыть ни одно из измерений:

def suggest_partition_transform(
    dominant_predicate_type: str,       # "range" | "equality" | "mixed"
    column_role: str,                   # "timestamp" | "high_cardinality_id" | "low_cardinality_category"
    expected_daily_volume_gb: float,
    concurrent_writers: int = 1,
) -> str:
    """
    Эвристический помощник для ОБСУЖДЕНИЯ дизайна партиционирования -
    не заменяет архитектурное решение, а структурирует критерии
    из раздела "Выбор Transform: практические критерии".
    """
    suggestions = []

    if column_role == "timestamp":
        if expected_daily_volume_gb < 1:
            suggestions.append("months(col) - дневной объём слишком мал для days()")
        elif expected_daily_volume_gb < 50:
            suggestions.append("days(col) - стандартный выбор для журналов событий")
        else:
            suggestions.append("hours(col) - высокий объём требует более гранулярного разбиения")

    if column_role == "high_cardinality_id" and dominant_predicate_type in ("equality", "mixed"):
        suggested_buckets = 16 if concurrent_writers <= 4 else 32
        suggestions.append(f"bucket({suggested_buckets}, col) - equality-pruning + защита от hotspot")

    if column_role == "low_cardinality_category":
        suggestions.append("identity(col) - низкая кардинальность, прямое значение")

    if concurrent_writers > 4 and column_role == "timestamp":
        suggestions.append(
            "Рассмотрите ДОБАВЛЕНИЕ bucket() как второго измерения "
            "для снижения конкуренции commit'ов между writer'ами"
        )

    return " + ".join(suggestions) if suggestions else "Недостаточно данных для рекомендации"


print(suggest_partition_transform(
    dominant_predicate_type="mixed",
    column_role="timestamp",
    expected_daily_volume_gb=80,
    concurrent_writers=6,
))
# hours(col) - высокий объём требует более гранулярного разбиения +
# Рассмотрите ДОБАВЛЕНИЕ bucket() как второго измерения для снижения
# конкуренции commit'ов между writer'ами

Hidden Partitioning и BI-инструменты: почему это больше, чем удобство для SQL¶

Преимущество «чистой» схемы без синтетических колонок особенно заметно за пределами прямых SQL-запросов аналитиков - в инструментах, которые подключаются к таблице через стандартный JDBC/ODBC драйвер и автоматически строят свой собственный слой метаданных (semantic layer) поверх схемы источника.

# То, что видит ЛЮБОЙ JDBC-клиент (включая BI-инструменты), подключаясь
# к Iceberg-таблице через Spark Thrift Server или эквивалентный сервис
import jaydebeapi  # иллюстративный пример JDBC-подключения

conn = jaydebeapi.connect(
    "org.apache.hive.jdbc.HiveDriver",
    "jdbc:hive2://spark-thrift-server:10000/default",
)
cursor = conn.cursor()
cursor.execute("DESCRIBE lakehouse.analytics.events_v2")
for row in cursor.fetchall():
    print(row)

# event_id    bigint
# user_id     bigint
# event_type  string
# event_ts    timestamp
# (партиционных полей нет - BI-инструмент строит свою модель
#  ИСКЛЮЧИТЕЛЬНО на основе этих четырёх бизнес-колонок)

Для BI-платформ, которые автоматически генерируют фильтры интерфейса на основе схемы таблицы (типичный паттерн в self-service аналитике), это устраняет целый класс проблем: пользователь интерфейса видит только event_ts как доступный для фильтрации временной атрибут, а не выбор между event_ts, event_date, event_year, event_month - что было бы прямым следствием классического Hive-партиционирования, где все синтетические колонки видны и потенциально доступны для выбора в интерфейсе, создавая путаницу («какую из четырёх похожих колонок выбрать для фильтра по дате?») и риск того, что пользователь интерфейса выберет «неправильную» с точки зрения partition pruning колонку, что в Hive-модели прямо ведёт к налогу на чтение, разобранному в начале урока, а в Iceberg - невозможно в принципе, потому что выбора между несколькими похожими колонками просто не существует.

Производственный кейс: смешанная нагрузка и комбинированное партиционирование¶

Ситуация. Платформа аналитики мобильного приложения хранила таблицу analytics.app_events (события взаимодействия пользователей с приложением) партиционированную только по days(event_ts) - решение, принятое на старте проекта, когда основная нагрузка были еженедельные batch-отчёты с фильтром по диапазону дат. Через год после запуска появился второй, не предусмотренный изначально паттерн использования: служба поддержки получила доступ к той же таблице для расследования жалоб конкретных пользователей - типичный запрос выглядел как WHERE user_id = X ORDER BY event_ts DESC LIMIT 100, без фильтра по дате вообще (служба поддержки не знала, когда произошёл инцидент).

Симптомы. Запросы службы поддержки систематически читали всю таблицу - несколько терабайт данных за полную историю - чтобы найти события одного конкретного пользователя, потому что единственное партиционирование (days(event_ts)) не давало никакого pruning без условия на дату. Время ответа на тикеты поддержки доходило до 5-10 минут на один запрос, что создавало заметное трение в работе службы поддержки и недовольство со стороны менеджеров, отвечающих за SLA обработки обращений.

Решение.

# Шаг 1: создаём новую версию таблицы с комбинированным партиционированием -
# монотонный transform для исторического паттерна (batch-отчёты)
# + bucket для нового паттерна (точечный поиск по user_id)
spark.sql("""
    CREATE TABLE lakehouse.analytics.app_events_v2 (
        event_id BIGINT, user_id BIGINT, event_type STRING, event_ts TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(event_ts), bucket(32, user_id))
""")

# Шаг 2: миграция исторических данных одной операцией -
# CTAS автоматически распределяет существующие данные
# по НОВОЙ комбинированной партиционной схеме
spark.sql("""
    INSERT INTO lakehouse.analytics.app_events_v2
    SELECT event_id, user_id, event_type, event_ts
    FROM lakehouse.analytics.app_events
""")

Результат. Запросы службы поддержки вида WHERE user_id = X (без фильтра по дате) после миграции читали в среднем 1/32 от объёма таблицы вместо 100% - prямое следствие equality-проекции предиката на bucket(32, user_id), отбрасывающей 31 из 32 бакетов независимо от того, что диапазон дат не указан. Время ответа сократилось с 5-10 минут до 15-25 секунд. При этом исторические batch-отчёты с фильтром по диапазону дат не показали никакой деградации - days(event_ts) продолжает работать ровно так же эффективно, как и раньше, потому что добавление bucket() как второго партиционного поля не отменяет pruning по первому, оно лишь добавляет ортогональное измерение отсечения.

Ключевой вывод. Партиционная схема, спроектированная под изначально известный паттерн нагрузки, не обязана оставаться неизменной, когда появляется новый паттерн - но переход к новой схеме на классическом Hive-подходе означал бы полную миграцию с риском для существующих пайплайнов и BI-дашбордов, ссылающихся на старые партиционные колонки в путях. В Iceberg та же миграция - это CREATE TABLE с новой partition spec и одна INSERT ... SELECT, не требующая ни единого изменения в коде дашбордов службы поддержки или batch-отчётов, потому что оба продолжают фильтровать по тем же самым бизнес-колонкам, что и раньше.

Типичные заблуждения про Hidden Partitioning¶

«Hidden Partitioning означает, что у Iceberg-таблицы нет физических партиций вообще» - неверно. Партиции физически существуют как директории/префиксы в object storage, и каждый файл принадлежит ровно одной комбинации значений партиционных полей. «Скрытость» относится к тому, что эта структура не видна как часть схемы и не требуется в SQL-запросах - но она абсолютно реальна на уровне физического хранения и manifest-метаданных.

«Если я не указал PARTITIONED BY, таблица не партиционирована и теряет всю производительность» - неточно. Таблица без явного партиционирования продолжает иметь полное дерево метаданных (manifest list, manifest files со статистикой) - File-level Pruning (Этап 3 алгоритма из прошлого урока) продолжает работать по любым колонкам со статистикой, независимо от партиционирования. Просто manifest-level pruning (Этап 2) не получает дополнительной агрегированной статистики по партиционным полям - то есть теряется только один уровень оптимизации, а не вся архитектура Scan Planning.

«Чем больше partition transform полей, тем лучше pruning» - неверно за пределами разумного. Каждое дополнительное поле партиционной спецификации увеличивает число уникальных комбинаций партиций, а значит - потенциально число файлов и манифестов (см. раздел про выбор transform и связь с manifest merging из третьего урока модуля). Добавление третьего, четвёртого транформа «на всякий случай» без реального паттерна запросов, который выиграет от этого измерения, создаёт чистый накладной расход без выигрыша.

«bucket() даёт такое же ускорение, как hash-партиционирование в традиционной БД» - близко по идее, но не идентично по применению. В традиционных СУБД хэш-партиционирование часто используется именно для распределения нагрузки записи между узлами кластера. В Iceberg bucket() не распределяет нагрузку между вычислительными узлами напрямую - он определяет, в какой физический файл попадёт строка, что влияет на pruning при чтении (для equality) и на равномерность размера файлов при записи, но не на то, какой именно executor обрабатывает эту строку в рамках конкретного Spark job.

«write.distribution-mode не влияет на корректность, только на производительность записи» - в общем случае верно, но с оговоркой, о которой важно знать: при distribution-mode='none' и сильно неотсортированных входных данных fanout-запись может настолько вырасти в числе одновременно открытых файловых дескрипторов, что job завершится с ошибкой (превышение лимита открытых файлов), а не просто медленнее отработает. В этом смысле выбор режима распределения иногда определяет не «быстрее или медленнее», а «упадёт или не упадёт» при достаточно большом числе уникальных партиций на task.

«Hidden Partitioning - это уникальная возможность только Apache Iceberg» - не совсем точно как общее утверждение про весь рынок table format, хотя в материалах этого курса фокус сознательно сделан на Iceberg как наиболее зрелой и distinct реализации этой идеи. У других table format есть собственные, архитектурно отличающиеся механизмы достижения похожего эффекта (например, data skipping по статистике в transaction log Delta Lake) - детальное сравнение подходов разных форматов вынесено в финальный урок модуля про выбор формата, чтобы не размывать фокус этого урока на специфике именно Iceberg-реализации.

«Раз партиционные значения не хранятся в самих файлах, их можно безопасно изменить без последствий» - неверно. Хотя физический Parquet-файл не содержит партиционных колонок (как было показано в разделе про Write Path), значение партиции каждого файла зафиксировано в манифесте на момент записи и неразрывно связано с конкретным partition-spec-id. Изменение того, как вычисляется партиция для уже записанных файлов, невозможно без переписывания этих файлов - именно поэтому Partition Evolution (следующий урок) работает только с новыми записями, оставляя историю нетронутой, а не «переинтерпретирует» старые файлы под новую формулу.

«Несколько partition transform полей всегда независимы друг от друга при pruning» - в основном верно для отбора манифестов/файлов (каждое поле проверяется отдельно и результаты комбинируются через AND, как было показано в разделе про составные спецификации), но не для интерпретации диапазонов: широкий диапазон по одному полю (например, bucket()-полю, которое почти не сужается) не компенсируется узким диапазоном по другому полю автоматически - оба поля вносят независимый вклад, и слабый pruning по одному измерению не «улучшается» только потому, что другое измерение партиционирования сильное.

Мостик к следующим урокам¶

Этот урок сфокусирован на одном конкретном, неизменном partition spec - следующие уроки модуля развивают эту тему в двух направлениях:

• Урок 5 («Partition Evolution») отвечает на вопрос, прямо вытекающий из производственного кейса этого урока: что происходит, если нужно изменить партиционную схему без полной миграции через CREATE TABLE + INSERT ... SELECT, сохранив историю снапшотов и не трогая уже записанные файлы. Тот же field-id, начинающийся с 1000 и разобранный в этом уроке, - основа механизма, который позволяет нескольким спецификациям партиционирования сосуществовать одновременно для разных частей истории одной таблицы.

• Уроки 6 и 7 («Copy-on-Write» и «Merge-on-Read») возвращаются к partition spec в контексте UPDATE/DELETE/MERGE - операции, которые должны знать текущую спецификацию партиционирования, чтобы корректно переписать или создать delete-файлы в правильных физических местах.

• Урок 13 («Выбор формата») сравнивает Hidden Partitioning Iceberg с аналогичными (но архитектурно иными) механизмами Delta Lake (partition columns остаются явной частью схемы, хотя Delta тоже поддерживает data skipping через статистику в transaction log) и Hudi.

• Урок 10 («Table Maintenance») возвращается к проблеме fanout-записи и мелких файлов с другой стороны - как обнаружить и исправить уже накопленные последствия неудачно сконфигурированного write.distribution-mode через rewrite_data_files, если проблема не была предотвращена на этапе записи.

Домашнее задание¶

1. Спроектируйте DDL для таблицы логов высокой интенсивности (представьте систему с десятками тысяч событий в секунду от множества независимых сервисов). Обоснуйте письменно выбор конкретной комбинации transform (например, hours(event_ts) + bucket(64, service_id)), опираясь на критерии из раздела «Выбор Transform: практические критерии».

2. Повторите Кейс 3 практического блока (сравнение Iceberg vs Hive-style) на собственных данных, но усложните сравнение: добейтесь ситуации, где аналитик случайно фильтрует по правильной колонке, но в неверном формате (например, WHERE event_ts >= 1718480400 - Unix-timestamp вместо строки даты). Проверьте, продолжает ли Predicate Projection работать в этом случае, и письменно объясните результат.

3. Напишите PySpark-скрипт, который для нескольких типов фильтров (точное равенство по bucket()-полю, диапазон по монотонному temporal-полю, равенство по полю без партиционирования, диапазон по bucket()-полю) выполняет запрос, извлекает Manifests skipped/Data Files skipped из плана выполнения, и сводит результат в таблицу - наглядно подтверждающую теоретическую матрицу «тип трансформа × тип предиката» из этого урока.

4. На таблице с комбинированным партиционированием days(event_ts), bucket(16, user_id) выполните INSERT с явным указанием диапазона user_id, заведомо избегающим один конкретный бакет (например, отфильтруйте при генерации тестовых данных так, чтобы bucket(16, user_id) != 5 для всех строк). Затем выполните запрос с фильтром на этот «пропущенный» бакет и убедитесь, что Data Files scanned = 0 - письменно объясните, почему в этом случае manifest-level pruning отбрасывает все манифесты без единого открытого файла.

5. Изучите (по документации Apache Iceberg или по экспериментам с вашей тестовой таблицей) поведение truncate(W, col) для отрицательных целых чисел и для значений NULL. Письменно (1-2 абзаца) опишите, как это поведение влияет на корректность Predicate Projection для предикатов вида col < 0 и col IS NULL.

6. Повторите Кейс 4 (измерение эффекта write.distribution-mode) с тремя значениями режима (none, hash, range) на одних и тех же неотсортированных данных. Сравните не только число итоговых файлов, но и scan time при последующем чтении с фильтром по партиционным полям - объясните, связано ли число файлов из Кейса 4 с эффективностью последующего Scan Planning, разобранного в третьем уроке модуля.

7. Создайте таблицу с явным NULL-значением в партиционируемой колонке (часть строк с event_ts IS NULL). Выполните SELECT * FROM <table>.manifests и найдите запись с contains_null = true в partition_summaries. Затем выполните два запроса - WHERE event_ts IS NULL и WHERE event_ts IS NOT NULL - и сравните Manifests skipped для каждого, объяснив результат через механизм, разобранный в разделе про NULL-значения и Predicate Projection.

8. Используя функцию suggest_partition_transform из раздела «Выбор Transform» как отправную точку, расширьте её одним дополнительным критерием на ваш выбор (например, учёт ожидаемого числа уникальных значений высококардинальной колонки для подбора числа бакетов N, а не фиксированных 16/32). Примените расширенную версию к реальной таблице из вашего проекта и сравните рекомендацию с тем, как таблица партиционирована сейчас.

Полная картина: от записи до чтения за один взгляд¶

Прежде чем перейти к итогам, полезно свести весь материал урока в единую сквозную диаграмму, объединяющую Write Path и Read Path - два процесса, разобранных в отдельных разделах, но связанных одной и той же partition spec.

Эта диаграмма - визуальное резюме главного архитектурного факта урока: Write Path и Read Path не знают друг о друге напрямую, но оба ссылаются на одну и ту же partition spec в metadata.json как на единственный источник истины о том, как устроено партиционирование. Именно эта общая точка ссылки - а не какая-либо синхронизация между записью и чтением в реальном времени - гарантирует, что данные, записанные сегодня, будут корректно прочитаны через pruning завтра, даже если между записью и чтением партиционная спецификация изменится (тема следующего урока).

Итоги¶

Hidden Partitioning превращает партиционирование из части логического контракта таблицы в деталь физической реализации. Аналитик фильтрует по бизнес-колонкам; инженер пишет чистые данные без синтетических колонок; партиционная функция (transform) - единственное место, где материализуется связь между ними, и она хранится в метаданных, а не в коде приложений.

Шесть встроенных transform решают разные, не взаимозаменяемые задачи. identity() и временные трансформы (year/month/day/hour) сохраняют порядок и эффективны для диапазонных предикатов; bucket(N) распределяет нагрузку равномерно через хэш и эффективен только для equality; truncate(W) даёт иерархическую группировку по префиксу, сохраняя монотонность.

Predicate Projection - конкретный алгоритм, а не магия. Iceberg знает формулу трансформа из partition spec и применяет её к границам предиката пользователя, получая эквивалентный предикат в домене значений партиции - именно этот спроецированный предикат сравнивается с агрегированной статистикой манифеста (алгоритм из третьего урока модуля). Для монотонных трансформов проекция диапазонных предикатов точна; для bucket() - только проекция equality-предикатов имеет смысл.

Партиционные значения никогда не попадают в сами файлы данных. Физический Parquet-файл, открытый напрямую через любой инструмент, минующий Iceberg, содержит только исходные бизнес-колонки - партиция существует исключительно как метаданные о расположении файла, что и делает возможным полное отсутствие синтетических колонок в схеме.

Выбор transform - архитектурное решение под конкретный паттерн нагрузки, не универсальный рецепт. Производственный кейс этого урока показал, что смешанная нагрузка (batch-отчёты по диапазону дат + точечный поиск по ID) требует комбинированного партиционирования, и переход к новой схеме в Iceberg не требует миграции кода потребителей данных - тема, которая получит полное развитие в следующем уроке про Partition Evolution.

Hidden Partitioning не устраняет ответственность инженера за качество записи - она перемещает её на другой уровень. Fanout-проблема и необходимость управлять write.distribution-mode показывают, что «скрытость» партиционирования снимает нагрузку с аналитика и с автора схемы запросов, но не освобождает инженера, проектирующего ETL-пайплайн, от понимания того, как именно данные физически распределяются по файлам при записи.

NULL-значения в партиционируемой колонке обрабатываются как полноправный случай, а не как исключение. Отдельный флаг contains_null в статистике манифеста, не зависящий от числового диапазона, позволяет Predicate Projection корректно обрабатывать IS NULL/IS NOT NULL без специальных директорий-заглушек, характерных для классического Hive-партиционирования.

Краткий глоссарий терминов урока¶

• Hidden Partitioning - механизм, при котором партиционная схема таблицы определяется через transform-функции от обычных колонок, а не через отдельные, видимые в схеме партиционные колонки.

• Partition Transform - функция, вычисляющая значение партиции из исходной колонки; шесть встроенных вариантов: identity, year, month, day, hour, bucket(N), truncate(W).

• Налог на чтение (Read Tax) - риск полного сканирования таблицы из-за фильтра на исходной колонке вместо синтетической партиционной колонки в классическом Hive-партиционировании.

• Налог на запись (Write Tax) - необходимость вручную вычислять и поддерживать синтетические партиционные колонки в ETL-коде при классическом Hive-партиционировании.

• Predicate Projection - алгоритм проекции границ предиката пользователя в домен значений partition transform, делающий возможным manifest-level pruning без явного упоминания партиции в запросе.

• Monotonic Transform - transform, сохраняющий относительный порядок исходных значений (identity, временные трансформы, truncate); допускает точную проекцию диапазонных предикатов.

• bucket(N, col) - хэш-transform, распределяющий значения по N бакетам через Murmur3-хэш с маскированием знакового бита и взятием остатка от деления; не сохраняет порядок, эффективен только для equality-предикатов.

• field-id (партиционного поля) - уникальный идентификатор поля partition spec, зарезервированный в диапазоне от 1000, отдельном от ID обычных колонок схемы.

• source-id - ссылка на ID исходной колонки схемы внутри определения партиционного поля, использующая тот же ID-based механизм, что и schema evolution.

• Fanout Write - паттерн записи, при котором task держит одновременно открытыми писатели для множества уникальных партиций из-за неотсортированного входного порядка строк; источник OOM и ошибок превышения лимита файловых дескрипторов.

• Clustered Write - паттерн записи, при котором строки сгруппированы по значению партиции перед записью, что позволяет держать открытым только один писатель за раз.

• write.distribution-mode - свойство таблицы, управляющее тем, выполняет ли Spark shuffle (hash/range) перед записью для гарантии clustered write, или пишет данные «как есть» (none).

• contains_null - булево поле в partition_field_summary манифеста, отдельно от числового диапазона lower_bound/upper_bound, отражающее наличие NULL-значений исходной колонки среди файлов манифеста; используется для проекции предикатов IS NULL/IS NOT NULL.

• CLUSTERED BY (Hive) - устаревший механизм классического Hive для хэш-бакетирования файлов внутри партиции без гарантий корректности при ручной записи; концептуальный предшественник bucket() в Iceberg, но без статуса полноценного partition-поля и без встроенной проверки инварианта.

• Налог на чтение / Налог на запись - два разных типа операционных издержек классического Hive-партиционирования: ответственность аналитика знать физическую структуру партиций (чтение) и необходимость инженера вручную поддерживать синтетические колонки (запись).

• identity(col) - простейший partition transform, где значение партиции в точности равно значению исходной колонки без преобразования; прямой аналог классического Hive-партиционирования, но без дублирования колонки в схеме.

• truncate(W, col) - partition transform, усекающий значение до первых W символов (строки) или округляющий вниз до кратного W (числа); сохраняет монотонность исходного порядка, в отличие от bucket().

• Semantic Layer (BI-инструмент) - слой метаданных, который self-service аналитические платформы строят автоматически на основе схемы источника данных; в случае Hidden Partitioning видит только бизнес-колонки, без синтетических партиционных полей.

• Hotspot (запись) - ситуация, при которой несколько конкурентных writer'ов одновременно пишут в один и тот же узкий диапазон партиций (например, текущий час), создавая повышенную частоту конфликтов commit'а; снижается добавлением bucket() как ортогонального измерения партиционирования.

• Bucket Map Join (Hive) - оптимизация JOIN в классическом Hive, использующая совпадающее бакетирование обеих сторон соединения для избежания полного shuffle; концептуальный, но не архитектурный предшественник bucket() в Iceberg.

• PartitionFilters (Spark physical plan) - секция в плане выполнения Spark, показывающая, какие условия запроса были применены как фильтр непосредственно на уровне партиций источника; пустая секция (PartitionFilters: []) для Hive-style таблицы - прямой индикатор отсутствия partition pruning для данного запроса.

• DESCRIBE TABLE EXTENDED - SQL-команда, показывающая дополнительную секцию # Partitioning с физической схемой partition spec, скрытую от обычного DESCRIBE TABLE, который отражает только бизнес-схему таблицы.

• suggest_partition_transform - вспомогательная функция-чек-лист из практической части урока, структурирующая критерии выбора transform (тип предиката, роль колонки, ожидаемый объём, число конкурентных writer'ов) в одну точку принятия решения на этапе проектирования таблицы.

• murmur3_32 - неконкурентная хэш-функция с равномерным распределением, используемая Iceberg как основа формулы bucket(N, col); не сохраняет порядок исходных значений, что и определяет границы применимости bucket-трансформа для pruning.

• Predicate Projection: точная vs ослабленная проекция - различие между точной проекцией для монотонных трансформов (диапазон сохраняется без потерь) и невозможностью смысловой проекции диапазона для bucket(), где сохраняется только проекция equality-предикатов.

• Партиционная директория-заглушка (__HIVE_DEFAULT_PARTITION__) - механизм классического Hive для представления NULL-значений в физическом пути партиции; в Iceberg заменён явным булевым флагом contains_null в статистике манифеста, не требующим специального значения-маркера в пути файла.

• PartitionSpec (объект) - программная репрезентация набора полей (source-id, transform, field-id, name), определяющих, как вычисляется партиция для каждой строки таблицы на конкретной версии спецификации.

• Equality-предикат vs Range-предикат - два класса условий фильтрации (col = X против col > X/BETWEEN), требующих разных по природе механизмов проекции и отличающихся по совместимости с монотонными и хэш-трансформами.

• CTAS-миграция партиционирования - паттерн перехода на новую партиционную схему через CREATE TABLE с новым PARTITIONED BY и последующий INSERT ... SELECT из старой таблицы, не требующий изменений в коде потребителей данных, поскольку обе таблицы предоставляют идентичную бизнес-схему.

• Self-Service Аналитика - модель работы с данными, где аналитики и нетехнические пользователи самостоятельно строят запросы и дашборды через BI-инструменты без участия инженеров; Hidden Partitioning снижает риск ошибок в этой модели, устраняя зависимость корректности фильтра от знания физического layout.

• Open File Handles Limit (ОС) - ограничение операционной системы на число одновременно открытых файловых дескрипторов процессом; типичная причина сбоя executor'а при fanout-записи в таблицу с большим числом уникальных партиций без распределения.

• Hidden Partitioning как карта территории урока - центральная идея: схема данных и схема партиционирования - две независимые сущности, связанные единственно через partition spec, без дублирования в физических файлах.

• Граница ответственности Write Path / Read Path - оба процесса ссылаются на одну и ту же partition spec из metadata.json, но не взаимодействуют друг с другом напрямую - это и обеспечивает корректность pruning независимо от того, когда именно были записаны конкретные файлы.