collect() и toPandas(): когда весь датасет едет в Driver

# ❌ Паттерн 1: collect() всего датасета
all_rows = df.collect()           # 1 TB → Driver → OOM
for row in all_rows:
    process(row)

# ❌ Паттерн 2: toPandas() без фильтрации
pd_df = df.toPandas()             # весь Spark DF → одна pandas таблица в памяти Driver

# ❌ Паттерн 3: collect() в цикле
for date in dates:
    result = df.filter(f"date='{date}'").collect()  # N запросов × collect()

# ❌ Паттерн 4: collect() для проверки "есть ли данные"
if df.collect():   # полный скан → Driver ради одного boolean
    ...

# ✅ Вместо collect() - write напрямую в хранилище
df.write.parquet("s3://bucket/output/")

# ✅ Если нужен sample - limit() + collect()
sample = df.limit(1000).collect()   # только 1000 строк

# ✅ Вместо collect() для проверки - isEmpty() или count()
df.isEmpty()       # O(1) - только проверяет есть ли партиции
df.count() == 0    # полный скан, но безопасен для Driver

# ✅ Вместо toPandas() - Arrow + limit
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
pd_df = df.limit(100_000).toPandas()   # Arrow + ограниченный объём

# ✅ Вместо цикла с collect() - используйте Spark трансформации
df.groupBy("date").agg(...)   # весь расчёт внутри кластера

# ❌ Огромная broadcast-переменная
lookup = load_giant_dict()    # 2 ГБ словарь
bc = spark.sparkContext.broadcast(lookup)
# Driver держит копию + шлёт на каждый Executor → 2 ГБ × N Executors

# ✅ Альтернатива - Broadcast Join через DataFrame
small_df = spark.createDataFrame(...)
df.join(broadcast(small_df), "key")   # Catalyst управляет размером

# Лимит broadcast через конфиг (default: spark.driver.maxResultSize = 1g)
spark.conf.set("spark.driver.maxResultSize", "2g")
# Превышение → SparkException: Job aborted due to stage failure

# ❌ Аккумулятор как хранилище данных
acc = spark.sparkContext.accumulator([], ListParam)
df.foreach(lambda r: acc.add(r))   # Все строки → Driver через accumulator

# ✅ Аккумулятор только для метрик
error_count = spark.sparkContext.accumulator(0)
def process(row):
    if row.value is None:
        error_count.add(1)
df.foreach(process)
print(f"Errors: {error_count.value}")

# Настроить Driver memory (только при spark-submit, не в коде)
# spark-submit --driver-memory 8g

# Лимит результата, который Driver принимает:
spark.conf.set("spark.driver.maxResultSize", "2g")  # default 1g