Schema Anti-Patterns: inferSchema в продакшне, вложенные структуры и from_json без схемы

# ❌ inferSchema = полное сканирование файла для определения типов
df = spark.read.option("inferSchema", "true").csv("s3://data/events.csv")
# Spark читает ВСЕ данные дважды:
# 1. Проход для inferSchema
# 2. Проход для реального чтения
# На 1 TB CSV - 2 TB чтения вместо 1 TB

# Хуже: тип int → long, float → double выбирается по максимальному значению
# "123" в одной строке, "1234567890123" в другой → Long для всей колонки
# Изменение данных → изменение схемы → падение downstream-кода

# ✅ Явная схема всегда
from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType

schema = StructType([
    StructField("user_id",    LongType(),   nullable=False),
    StructField("event_type", StringType(), nullable=True),
    StructField("amount",     DoubleType(), nullable=True),
    StructField("event_ts",   LongType(),   nullable=True),
])

df = spark.read.schema(schema).csv("s3://data/events.csv")
# Одно чтение, типы определены заранее, схема задокументирована в коде

# ✅ Для Parquet - схема хранится в файле, inferSchema безопасен
# (Parquet самоописывающий формат)
df = spark.read.parquet("s3://data/events.parquet")   # OK - схема из метаданных

# ❌ schema_of_json инфериит схему из одной строки
sample = '{"user_id": 1, "tags": ["a", "b"]}'
inferred_schema = schema_of_json(sample)

df = df.withColumn("parsed", from_json("json_col", inferred_schema))
# Проблемы:
# - sample может не представлять все возможные поля
# - типы угадываются: "1" → LongType, но в других строках может быть "1.5"
# - null-поля в sample → StructField с NullType → всё null после парсинга

# ✅ Явная схема JSON
from pyspark.sql.types import *

json_schema = StructType([
    StructField("user_id",  LongType(),  True),
    StructField("email",    StringType(), True),
    StructField("tags",     ArrayType(StringType()), True),
    StructField("metadata", MapType(StringType(), StringType()), True),
])

df.withColumn("parsed", from_json("json_col", json_schema))

# ❌ Тройная вложенность - nightmare для разработчика и для Spark
df = spark.read.json("s3://events/")
# Схема: event.context.device.os.version - 4 уровня!

# Доступ к глубокому полю:
df.select("event.context.device.os.version")   # Если хоть одно поле null - NPE

# Column pruning работает только до определённой глубины
# Весь вложенный struct читается даже если нужно одно поле

# ✅ Flatten при чтении - один раз, на входе
from pyspark.sql.functions import col

df_flat = df.select(
    col("event.id").alias("event_id"),
    col("event.type").alias("event_type"),
    col("event.context.device.os.version").alias("os_version"),
    col("event.context.session_id").alias("session_id"),
)
# Плоская схема: быстрее, понятнее, лучше pushdown

# ❌ from_json с разной схемой для разных строк - невозможно векторизовать
@udf("string")
def parse_dynamic(json_str, event_type):
    data = json.loads(json_str)
    if event_type == "purchase":
        return data.get("amount")
    else:
        return data.get("duration")
# Весь JSON парсится Python построчно, нет Arrow оптимизации

# ✅ Единая схема + when/otherwise
from pyspark.sql.functions import from_json, when

unified_schema = StructType([
    StructField("amount",   DoubleType(), True),
    StructField("duration", IntegerType(), True),
])

df.withColumn("parsed",  from_json("json_col", unified_schema)) \
  .withColumn("value",
    when(col("event_type") == "purchase", col("parsed.amount"))
    .otherwise(col("parsed.duration").cast("double"))
  )

# ❌ Новая колонка добавлена в Parquet-файл, старый код падает
df = spark.read.parquet("s3://data/")   # new_column появилась в новых файлах
# Результат: AnalysisException - поле отсутствует в части файлов
# или: новое поле заполнено null без предупреждения

# ✅ Явно включить schema merging
df = spark.read \
    .option("mergeSchema", "true") \
    .parquet("s3://data/")
# Spark объединяет схемы всех файлов, новые колонки заполняются null

# ✅ Для Iceberg - schema evolution встроена в таблицу
spark.sql("ALTER TABLE events ADD COLUMN new_field STRING")

# Всегда проверять схему после чтения
df.printSchema()    # дерево типов
df.dtypes           # список (name, type)

# Искать подозрительные типы:
# StringType там, где должен быть IntegerType → вероятно inferSchema промахнулся
# NullType → поле есть в схеме, но всегда null

# Проверить null-distribution
from pyspark.sql.functions import isnan, when, count, col
null_counts = df.select([
    count(when(col(c).isNull(), c)).alias(c) for c in df.columns
])
null_counts.show()