Parquet на S3: vectorized reader, column pruning и row group filtering

┌─────────────────────────────────────────────────────────────────┐
│  Magic Number: PAR1 (4 байта)                                   │
├─────────────────────────────────────────────────────────────────┤
│  ROW GROUP 1 (128–512 MB по умолчанию)                         │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │  Column Chunk: user_id                                   │   │
│  │  ┌────────────┐ ┌────────────┐ ┌────────────┐           │   │
│  │  │  Page 1    │ │  Page 2    │ │  Page 3    │ ...        │   │
│  │  │ (8-16KB)   │ │ (8-16KB)  │ │ (8-16KB)   │           │   │
│  │  └────────────┘ └────────────┘ └────────────┘           │   │
│  │  Column Chunk: amount                                    │   │
│  │  ┌────────────┐ ┌────────────┐ ...                       │   │
│  │  └────────────┘ └────────────┘                           │   │
│  │  Column Chunk: event_date                                │   │
│  │  ┌────────────┐ ...                                      │   │
│  └──────────────────────────────────────────────────────────┘   │
├─────────────────────────────────────────────────────────────────┤
│  ROW GROUP 2                                                    │
│  ...                                                            │
├─────────────────────────────────────────────────────────────────┤
│  ROW GROUP N                                                    │
│  ...                                                            │
├─────────────────────────────────────────────────────────────────┤
│  FOOTER (переменный размер, обычно КБ–МБ)                       │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │  Схема данных (Schema)                                   │   │
│  │  Для каждой Row Group:                                   │   │
│  │    Для каждой колонки:                                   │   │
│  │      - byte_offset: где в файле начинается Column Chunk  │   │
│  │      - total_compressed_size                             │   │
│  │      - statistics: min/max/null_count                    │   │
│  └──────────────────────────────────────────────────────────┘   │
│  Footer Length (4 байта)                                       │
│  Magic Number: PAR1 (4 байта)                                   │
└─────────────────────────────────────────────────────────────────┘

SELECT user_id, SUM(amount) as total
FROM events
WHERE event_date = '2024-06-15'
GROUP BY user_id

from pyspark.sql import SparkSession
from pyspark.sql import functions as F


spark = SparkSession.builder.appName("parquet-demo").getOrCreate()

# Читаем таблицу с 200 колонками (10 GB):
events = spark.read.parquet("s3a://data-lake/events/")

# ❌ SELECT * - Column Pruning не работает, читаем все 200 колонок:
events.select("*") \
    .filter(F.col("event_date") == "2024-06-15") \
    .groupBy("user_id") \
    .agg(F.sum("amount")) \
    .show()

# ✅ Точечный SELECT - Column Pruning читает 3 из 200 колонок:
events.select("user_id", "amount", "event_date") \
    .filter(F.col("event_date") == "2024-06-15") \
    .groupBy("user_id") \
    .agg(F.sum("amount")) \
    .show()

# ✅ Spark автоматически применяет Column Pruning даже без explicit select:
# Catalyst видит, что groupBy/agg использует только user_id и amount,
# а filter использует только event_date → автоматически запрашивает 3 колонки
events \
    .filter(F.col("event_date") == "2024-06-15") \
    .groupBy("user_id") \
    .agg(F.sum("amount")) \
    .show()

# Посмотреть Physical Plan:
events \
    .filter(F.col("event_date") == "2024-06-15") \
    .groupBy("user_id") \
    .agg(F.sum("amount")) \
    .explain(extended=True)

# В выводе ищем ReadSchema в операторе FileScan:
# == Physical Plan ==
# ...
# +- *(1) FileScan parquet [user_id#0L, amount#1, event_date#2]
#    Batched: true
#    Location: InMemoryFileIndex(...)
#    PushedFilters: [IsNotNull(event_date), EqualTo(event_date,2024-06-15)]
#    ReadSchema: struct<user_id:bigint, amount:double, event_date:date>
#
# ReadSchema содержит только 3 колонки из 200 - Column Pruning работает!

# Схема: events.metadata STRUCT<device: STRUCT<type: STRING, os: STRING>, app_version: STRING>
# Нужно только metadata.device.type:

events.select(
    "user_id",
    F.col("metadata.device.type").alias("device_type")
).show()

# Physical Plan покажет:
# ReadSchema: struct<user_id:bigint, metadata:struct<device:struct<type:string>>>
# Spark читает только поле type внутри device внутри metadata - не весь STRUCT!

from pyspark.sql.functions import udf
from pyspark.sql.types import DoubleType

# ❌ UDF принимает весь Row объект - Column Pruning отключается!
@udf(returnType=DoubleType())
def complex_calc(row):
    return row.amount * 1.1

# Spark не знает, какие поля использует UDF → читает все колонки
events.withColumn("adjusted_amount", complex_calc(F.struct("*"))).show()

# ✅ UDF получает только нужные колонки:
@udf(returnType=DoubleType())
def calc_adjusted_amount(amount: float) -> float:
    return amount * 1.1

# Spark знает: UDF нужна только колонка amount
events.withColumn(
    "adjusted_amount",
    calc_adjusted_amount(F.col("amount"))
).select("user_id", "adjusted_amount").show()

# Запрос с фильтром:
result = events \
    .filter(F.col("event_date") == "2024-06-15") \
    .filter(F.col("amount") > 100) \
    .select("user_id", "amount")

result.explain()

# Physical Plan (ключевые строки):
# +- *(1) Project [user_id#0L, amount#1]
#    +- *(1) Filter (isnotnull(amount#1) AND (amount#1 > 100.0))
#       +- *(1) FileScan parquet [user_id#0L, amount#1, event_date#2]
#          Batched: true
#          PushedFilters: [IsNotNull(event_date), EqualTo(event_date,2024-06-15),
#                          IsNotNull(amount), GreaterThan(amount,100.0)]
#          ReadSchema: struct<user_id:bigint, amount:double, event_date:date>

# ❌ Несортированные данные: фильтр event_date = '2024-06-15' неэффективен
# Каждая Row Group содержит случайный микс дат:
# RG1: min=2024-01-01, max=2024-12-31 (почти весь год)
# RG2: min=2024-01-02, max=2024-12-30
# RG3: min=2024-01-01, max=2024-12-31
# → Все Row Groups «могут содержать нужные строки» → читаем всё

# ✅ Отсортированные данные: каждая Row Group = узкий диапазон дат
events_sorted = events.sortWithinPartitions("event_date")
# RG1: min=2024-01-01, max=2024-01-31 (январь)
# RG2: min=2024-02-01, max=2024-02-28 (февраль)
# RG3: ...
# RG6: min=2024-06-01, max=2024-06-30 (июнь) ← только эта RG читается!

# Запись с правильной сортировкой:
events \
    .sortWithinPartitions("event_date", "user_id") \
    .write \
    .option("parquet.block.size", 134217728) \  # 128 MB Row Groups
    .parquet("s3a://data-lake/events-sorted/")

# Включить запись page-level statistics:
spark.conf.set("spark.sql.parquet.enablePageIndexing", "true")

# При чтении Spark использует Page Index для additional filtering:
# Вместо «читать весь Column Chunk» → «читать только страницы с нужными данными»
# Это особенно эффективно для частичных фильтров внутри Row Group

# Включить Bloom Filter при записи:
df.write \
    .option("parquet.bloom.filter.enabled", "true") \
    .option("parquet.bloom.filter.expected.ndv", "1000000") \  # ожидаемое кол-во уникальных значений
    .option("parquet.bloom.filter.fpp", "0.05") \              # false positive rate 5%
    .parquet("s3a://data-lake/users/")

# Spark использует Bloom Filter автоматически при наличии EqualTo фильтров:
spark.conf.set("spark.sql.parquet.enableBloomFilterPushdown", "true")

# Теперь: WHERE user_id = 12345
# → Spark проверяет Bloom Filter каждой RG (несколько KB из Footer)
# → Если Bloom Filter говорит «нет» → RG пропускается без чтения данных
# → Bloom Filter никогда не даёт false negative, только false positive

Batch size: 4096 строк
user_id:    int64[4096]  - плоский массив int64
amount:     float64[4096] - плоский массив float64
is_active:  bool[4096]   - плоский массив bool (или bit array)

# Включён по умолчанию в Spark 2.0+:
spark.conf.set("spark.sql.parquet.enableVectorizedReader", "true")

# Размер batch (строк за одну итерацию):
spark.conf.set("spark.sql.parquet.columnarReaderBatchSize", "4096")
# Увеличение до 8192 может ускорить простые запросы
# Уменьшение помогает при очень широких таблицах (много колонок)

# Максимальный размер Column Vector (в памяти):
# При batch=4096 и 200 колонок int64:
# 4096 × 200 × 8 bytes = 6.5 MB per batch - нормально

# Проверить, используется ли Vectorized Reader:
# В Physical Plan ищем "Batched: true":
# +- *(1) FileScan parquet [...]
#    Batched: true   ← Vectorized Reader активен
#    Batched: false  ← Row-by-Row Reader (деградация)

# ❌ Vectorized Reader отключается для ARRAY/MAP/глубоких STRUCT:
events.select(
    "user_id",
    F.explode("tags")  # tags: ARRAY<STRING>
).show()
# Physical Plan: Batched: false → Row-by-Row Reader

# ✅ Для простых STRUCT верхнего уровня Vector Reader работает:
events.select(
    "user_id",
    F.col("metadata.device_type")  # STRUCT с простым полем
).show()
# Physical Plan: Batched: true (если нет глубокой вложенности)

from pyspark.sql.functions import udf

# ❌ Python UDF требует сериализации в Python-объекты → no vectorized:
@udf("double")
def python_calc(x):
    return x * 1.1

events.withColumn("adj", python_calc("amount")).show()
# Batched: false - каждая строка сериализуется в Python

# ✅ Pandas UDF (Vectorized UDF) работает с Column Vectors:
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf("double")
def pandas_calc(series: pd.Series) -> pd.Series:
    return series * 1.1

events.withColumn("adj", pandas_calc("amount")).show()
# Batched: true - работает с pandas Series (векторизовано)

# Decimal с precision > 38 → Row-by-Row
# BINARY columns → Row-by-Row (в некоторых версиях Spark)
# Проверяйте через explain() + ищите Batched: true/false

# Размер Row Group в байтах:
spark.conf.set("spark.hadoop.parquet.block.size", str(256 * 1024 * 1024))  # 256 MB

# Или через write options:
df.write \
    .option("parquet.block.size", 268435456) \   # 256 MB
    .option("parquet.page.size", 1048576) \       # 1 MB страницы
    .option("parquet.dictionary.enabled", "true") \  # Dictionary encoding
    .parquet("s3a://data-lake/events-optimized/")

# Проверить фактический Row Group layout Parquet-файла:
from pyspark.sql import functions as F

spark.read \
    .parquet("s3a://data-lake/events-optimized/part-00000.parquet") \
    .select("*") \
    .explain()

# Или через parquet-tools CLI:
# parquet-tools meta part-00000.parquet | grep "row group"

# ❌ Плохо: 10 000 файлов по 1 MB
streaming_df.writeStream \
    .trigger(processingTime="10 seconds") \  # частые триггеры = много мелких файлов
    .parquet("s3a://bucket/events/")

# Посмотреть, сколько файлов:
spark.read.parquet("s3a://bucket/events/").inputFiles.__len__()
# → 10 000 файлов: 10 000 Footer GET запросов при каждом чтении

# ✅ Компакция через Spark:
def compact_parquet(
    spark,
    input_path: str,
    output_path: str,
    target_file_size_mb: int = 256,
    target_files: int = None
) -> None:
    """
    Объединить мелкие Parquet-файлы в крупные.
    target_files: None = авто (spark.sql.shuffle.partitions)
    """
    df = spark.read.parquet(input_path)

    if target_files is None:
        # Оценить оптимальное количество файлов:
        # total_size / target_file_size_mb
        # Spark не знает размер сам, используем shuffle.partitions
        target_files = max(1, spark.sparkContext.defaultParallelism)

    df.coalesce(target_files) \
        .sortWithinPartitions("event_date", "user_id") \  # сортировка для Row Group Filtering
        .write \
        .option("parquet.block.size", target_file_size_mb * 1024 * 1024) \
        .mode("overwrite") \
        .parquet(output_path)

# После компакции: 50 файлов по 256 MB
# 50 Footer GET запросов вместо 10 000 = в 200 раз меньше S3 API вызовов

# Iceberg с hidden partitioning по дате:
spark.sql("""
    CREATE TABLE catalog.events (
        user_id    BIGINT,
        event_type STRING,
        amount     DOUBLE,
        created_at TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (days(created_at))  -- hidden partition
""")

# Запрос с фильтром по времени:
spark.sql("""
    SELECT user_id, SUM(amount)
    FROM catalog.events
    WHERE created_at >= '2024-06-15'
      AND created_at <  '2024-06-16'
    GROUP BY user_id
""")

# Iceberg делает:
# 1. Читает Manifest List (1 HTTP GET)
# 2. Прочитает только Manifests для partition days=19889 (2024-06-15)
# 3. В пределах манифеста: проверяет min/max статистику каждого файла
# 4. Для выбранных файлов: Row Group Filtering из Parquet Footer
# → 3 уровня pruning вместо 1 (только Parquet Footer)

# Z-ordering данных по нескольким колонкам:
spark.sql("""
    OPTIMIZE delta.`s3a://data-lake/events`
    ZORDER BY (user_id, event_date)
""")

# После Z-ordering:
# Данные кластеризованы по (user_id, event_date) одновременно
# Запрос WHERE user_id = 12345 AND event_date = '2024-06-15':
# → Row Groups с нужным user_id И нужной датой сгруппированы вместе
# → Большинство Row Groups пропускается по min/max обоих полей
# → Работает для ЛЮБОГО подмножества Z-ordered колонок

# Аналог в Iceberg через Sort Order:
spark.sql("""
    ALTER TABLE catalog.events
    WRITE ORDERED BY user_id, event_date
""")
# Или через rewrite:
spark.sql("""
    CALL spark_catalog.system.rewrite_data_files(
        table => 'catalog.events',
        strategy => 'sort',
        sort_order => 'user_id ASC, event_date ASC'
    )
""")

# ❌ SELECT * отменяет Column Pruning:
events.select("*") \
    .filter("event_date = '2024-06-15'") \
    .count()
# Читает все 200 колонок из S3 → 10 GB трафика

# ✅ Точечный SELECT:
events.select("user_id", "amount", "event_date") \
    .filter("event_date = '2024-06-15'") \
    .count()
# Читает 3 колонки → ~150 MB трафика (67× меньше)

# ❌ Чтение обеих таблиц полностью, потом join и фильтрация:
orders = spark.read.parquet("s3a://data-lake/orders/")       # 500 GB
customers = spark.read.parquet("s3a://data-lake/customers/") # 50 GB

result = orders.join(customers, "customer_id") \
    .filter("orders.order_date = '2024-06-15'")  # фильтр применяется к результату join

# ✅ Предварительная фильтрация перед join:
orders_filtered = orders.filter("order_date = '2024-06-15'")  # Row Group Filtering работает
# → читает только нужные Row Groups: 5 GB вместо 500 GB

result = orders_filtered.join(customers, "customer_id")

# ❌ Фильтр по производному полю - Predicate Pushdown НЕ работает:
events \
    .withColumn("year", F.year("event_date")) \
    .filter(F.col("year") == 2024)
# PushedFilters: [] - пустой! Читается ВСЁ, потом фильтрация в памяти

# ✅ Фильтр по оригинальному полю:
events \
    .filter((F.col("event_date") >= "2024-01-01") & (F.col("event_date") < "2025-01-01"))
# PushedFilters: [GreaterThanOrEqual(event_date,2024-01-01), LessThan(event_date,2025-01-01)]

# ✅ Или через SQL (Catalyst обрабатывает умнее):
events.createOrReplaceTempView("events")
spark.sql("SELECT * FROM events WHERE YEAR(event_date) = 2024")
# Catalyst может преобразовать YEAR(...) = 2024 в диапазонный фильтр

# ❌ Партиционирование по user_id (миллионы уникальных значений):
events.write \
    .partitionBy("user_id") \          # 10М уникальных user_id = 10М директорий!
    .parquet("s3a://data-lake/events/")

# Результат:
# - 10М партиций = 10М ListObjects запросов для планирования
# - Каждая партиция: 1 крошечный файл в несколько KB
# - Spark Driver OutOfMemory на этапе листинга директорий

# ✅ Партиционирование по низкокардинальному полю + Row Group Filtering для user_id:
events.write \
    .partitionBy("event_date") \       # ~365 партиций за год
    .sortWithinPartitions("user_id") \ # Сортировка для Row Group Filtering
    .parquet("s3a://data-lake/events/")

# Запрос WHERE event_date='2024-06-15' AND user_id = 12345:
# 1. Partition pruning: читаем только event_date=2024-06-15 (1 из 365 директорий)
# 2. Row Group Filtering: пропускаем Row Groups где max(user_id) < 12345 или min > 12345
# 3. Bloom Filter (если включён): точечный поиск user_id внутри RG

# ❌ Отключение Dictionary Encoding для строковых колонок:
df.write \
    .option("parquet.enable.dictionary", "false") \  # Отключили dictionary!
    .parquet("s3a://data-lake/events/")

# Без dictionary каждое значение кодируется как raw string
# "purchase" (8 bytes) × 1M строк = 8 MB
# С dictionary: создаётся словарь {"purchase": 0, "view": 1, ...},
# каждое значение = 1 byte index → 1 MB (8× сжатие)

# ✅ Dictionary Encoding включён по умолчанию:
df.write \
    .option("parquet.enable.dictionary", "true") \   # по умолчанию true
    .option("parquet.dictionary.page.size", "1048576") \  # 1 MB словарь
    .parquet("s3a://data-lake/events/")

Таблица: s3a://data-lake/events/
Размер: 2 TB, 10 000 файлов, 200 колонок
Формат: Parquet, файлы ~200 MB, Row Groups ~128 MB
Данные: НЕ отсортированы, партиций нет

# Запрос ДО оптимизации (45 минут):
spark.read.parquet("s3a://data-lake/events/") \
    .filter(F.col("event_date") == "2024-06-15") \
    .filter(F.col("country") == "RU") \
    .groupBy("user_id") \
    .agg(F.sum("amount").alias("total")) \
    .show()

# Spark UI показывает:
# Scanned: 2 TB
# Rows output: 15,230
# Scan time: 42 minutes
# S3 GET requests: ~10,000 (footer) + ~200,000 (data)

# Проверяем Physical Plan:
spark.read.parquet("s3a://data-lake/events/") \
    .filter(F.col("event_date") == "2024-06-15") \
    .filter(F.col("country") == "RU") \
    .groupBy("user_id") \
    .agg(F.sum("amount").alias("total")) \
    .explain()

# ReadSchema: struct<user_id:bigint, amount:double, event_date:date, country:string>
# ✅ Column Pruning работает: 4 из 200 колонок

# Эффект: 2 TB → ~40 GB (50× меньше данных из S3 по объёму колонок)
# Но Row Groups пока не пропускаются (данные не отсортированы)
# Время: 45 min → ~12 min

# Переписать данные с партиционированием:
spark.read.parquet("s3a://data-lake/events/") \
    .write \
    .partitionBy("event_date") \       # 365 партиций за год
    .sortWithinPartitions("country", "user_id") \
    .option("parquet.block.size", 268435456) \  # 256 MB Row Groups
    .mode("overwrite") \
    .parquet("s3a://data-lake/events-v2/")

# Теперь запрос с event_date фильтром:
spark.read.parquet("s3a://data-lake/events-v2/") \
    .filter(F.col("event_date") == "2024-06-15") \
    .filter(F.col("country") == "RU") \
    .groupBy("user_id") \
    .agg(F.sum("amount").alias("total")) \
    .show()

# Partition pruning: читаем только event_date=2024-06-15 (1/365 данных)
# Column Pruning: 4 из 200 колонок
# Row Group Filtering по country (данные отсортированы по country внутри партиции)
# Эффект: ~40 GB → ~500 MB
# Время: 12 min → ~90 секунд

# Переписать с Bloom Filter на колонке country:
spark.read.parquet("s3a://data-lake/events-v2/") \
    .write \
    .partitionBy("event_date") \
    .sortWithinPartitions("country", "user_id") \
    .option("parquet.block.size", 268435456) \
    .option("parquet.bloom.filter.enabled", "true") \
    .option("parquet.bloom.filter.column.names", "country") \
    .option("parquet.bloom.filter.expected.ndv", "200") \  # ~200 стран
    .mode("overwrite") \
    .parquet("s3a://data-lake/events-v3/")

spark.conf.set("spark.sql.parquet.enableBloomFilterPushdown", "true")

# Запрос WHERE country = 'RU':
# → Bloom Filter мгновенно исключает Row Groups без записей RU
# → Ещё более точное пропускание по min/max (данные отсортированы)
# Время: 90 sec → ~54 секунды

def print_query_stats(df_plan_str: str, description: str):
    """Вывести ключевые метрики из Physical Plan."""
    print(f"\n{'='*50}")
    print(f"Версия: {description}")
    print(f"{'='*50}")
    print(df_plan_str)


# Метрики ДО → ПОСЛЕ оптимизации:
stats = {
    "S3 трафик":           "2 TB  →  ~500 MB    (4000× меньше)",
    "Scanned Row Groups":  "16 000 → ~60         (267× меньше)",
    "S3 GET requests":     "210 000 → ~800       (262× меньше)",
    "Scan time":           "42 мин → ~54 сек     (47× быстрее)",
    "S3 стоимость ($/GB)": "$46.00 → $0.012      (3800× дешевле)",
}

for metric, comparison in stats.items():
    print(f"  {metric}: {comparison}")

FileScan parquet:
├── number of files read: 27           ← сколько файлов прочитано
├── scan time total: 00:01:23          ← время чтения из S3
├── metadata time: 00:00:02            ← время на Footer запросы
├── number of output rows: 15,230      ← строки после Row Group Filtering
├── bytes scanned: 512,345,678         ← байты прочитанные из S3
└── files pruned: 9,973                ← файлы пропущены partition pruning