Глоссарий — Apache Spark Internals

// Пять свойств RDD в исходниках Spark:
// abstract class RDD[T] {
//   def getPartitions: Array[Partition]
//   def compute(split: Partition, ctx: TaskContext): Iterator[T]
//   def getDependencies: Seq[Dependency[_]]
//   val partitioner: Option[Partitioner]
//   def getPreferredLocations(split: Partition): Seq[String]
// }

val rdd = sc.textFile("hdfs:///data/*.log")  // HadoopRDD
val mapped = rdd.map(line => line.split(","))  // MapPartitionsRDD
val filtered = mapped.filter(_.length > 3)     // MapPartitionsRDD
// Граф зависимостей создан, данных ещё нет

// Визуализация в Spark UI:
// Job X:
//   Stage 0 (ShuffleMapStage): 200 tasks → shuffle write
//   Stage 1 (ShuffleMapStage): 200 tasks → shuffle write
//   Stage 2 (ResultStage):     100 tasks → write to parquet

// В логах DAGScheduler:
// INFO DAGScheduler: Submitting ShuffleMapStage 0 (200 tasks)
// INFO DAGScheduler: ShuffleMapStage 0 finished in 12.3 s
// INFO DAGScheduler: Submitting ResultStage 2 (100 tasks)

// Ключевое: если ShuffleMapStage упал частично
// → пересчитываются только потерянные map outputs
// → ResultStage перезапускается полностью

val rdd = sc.textFile("/data")
  .map(_.split(","))
  .filter(_(2).toInt > 100)
  .map(arr => (arr(0), arr(1).toLong))
  .reduceByKey(_ + _)

// Печать lineage
println(rdd.toDebugString)
// (8) ShuffledRDD[5]
//  +-(8) MapPartitionsRDD[4]
//      |  MapPartitionsRDD[3]
//      |  MapPartitionsRDD[2]
//      |  MapPartitionsRDD[1]
//      |  /data HadoopRDD[0]

// Narrow: каждый выходной partition ← один входной
val rdd2 = rdd1.map(x => x * 2)       // MapPartitionsRDD: narrow
val rdd3 = rdd2.filter(_ > 5)          // MapPartitionsRDD: narrow
val rdd4 = rdd1.union(rdd2)            // UnionRDD: narrow

// Конвейеризация: map + filter выполняются в одном task
// без записи промежуточных данных на диск

// Восстановление: потеря partition i в rdd3
// → пересчитать только partition i rdd1 → rdd2 → rdd3

// Wide: каждый выходной partition ← много входных
val counts = rdd.groupByKey()   // ShuffledRDD: wide
val joined = rdd1.join(rdd2)    // CoGroupedRDD: wide

// DAGScheduler создаёт stage-границу здесь:
// Stage 0: [upstream ops] → shuffle write
// Stage 1: shuffle read → [downstream ops]

// Восстановление wide dependency дорогостоящее:
// потеря partition требует повторного shuffle из Stage 0

import org.apache.spark.HashPartitioner
import org.apache.spark.RangePartitioner

// HashPartitioner — по умолчанию для groupByKey/reduceByKey
val partitioned = rdd.partitionBy(new HashPartitioner(100))

// RangePartitioner — для sortByKey
val rangePart = new RangePartitioner(100, rdd)  // сэмплирует данные!
val sorted = rdd.sortByKey()  // использует RangePartitioner

// Custom Partitioner
class DomainPartitioner(n: Int) extends Partitioner {
  def numPartitions: Int = n
  def getPartition(key: Any): Int =
    Math.abs(key.hashCode) % n  // или своя логика
}

// DAGScheduler не вызывается напрямую, но его поведение видно через:

// 1. Spark UI: вкладка Jobs → stages breakdown
// 2. Логи: [DAGScheduler] Submitting ShuffleMapStage 0
// 3. eventLog: SparkListenerStageSubmitted

// Ключевые методы (Scala internals):
// DAGScheduler.handleJobSubmitted() → создаёт finalStage
// DAGScheduler.getOrCreateShuffleMapStage() → для wide deps
// DAGScheduler.submitStage() → проверяет готовность родителей

// Конфигурация задержки для delay scheduling:
spark.conf.set("spark.locality.wait", "3s")         // общее ожидание
spark.conf.set("spark.locality.wait.node", "3s")     // NODE_LOCAL
spark.conf.set("spark.locality.wait.rack", "3s")     // RACK_LOCAL
// PROCESS_LOCAL → NODE_LOCAL → RACK_LOCAL → ANY

// Метрики в Spark UI → Stages → Task Metrics:
// Locality Level: PROCESS_LOCAL/NODE_LOCAL/RACK_LOCAL/ANY
// Наличие только ANY-задач → проблема с локальностью

// Настройки TaskSetManager:
spark.conf.set("spark.task.maxFailures", "4")  // макс. попыток
spark.conf.set(
  "spark.speculation", "true"
)  // включить speculative execution
spark.conf.set(
  "spark.speculation.multiplier", "1.5"
)  // задача-straggler: >1.5x медианы
spark.conf.set(
  "spark.speculation.quantile", "0.75"
)  // ждать 75% завершённых задач

// В логах:
// TaskSetManager: Starting task 5.0 in stage 2.0 (TID 15)
// TaskSetManager: Lost task 5.0 in stage 2.0 — retrying (1/4)

// Проверка локальности в Spark UI → Stage → Tasks:
// PROCESS_LOCAL: кешированный RDD/DataFrame в памяти executor
// NODE_LOCAL: блок HDFS на том же узле, executor не кеширует
// RACK_LOCAL: блок HDFS на другом узле того же rack
// ANY: данные на другом rack или object storage (S3/GCS)

// Для S3/GCS всегда ANY — нет данных на executor-нодах
// Оптимизация: увеличить cache, уменьшить locality.wait для S3
spark.conf.set("spark.locality.wait", "0s")  // для object storage

// Включение speculative execution
spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.interval", "100ms")  // частота проверки
spark.conf.set("spark.speculation.multiplier", "1.5")   // порог straggler
spark.conf.set("spark.speculation.quantile", "0.75")    // % завершённых задач

// В Spark UI → Stage → Tasks:
// task с (speculative) в поле Index — это дубликат
// Успех первого завершённого убивает второй

// ОСТОРОЖНО: non-idempotent задачи могут давать двойные записи!

spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.minExecutors", "2")
spark.conf.set("spark.dynamicAllocation.maxExecutors", "100")
spark.conf.set("spark.dynamicAllocation.initialExecutors", "5")
spark.conf.set("spark.dynamicAllocation.executorIdleTimeout", "60s")

// Для Kubernetes (без External Shuffle Service):
spark.conf.set(
  "spark.dynamicAllocation.shuffleTracking.enabled", "true"
)
// Spark отслеживает, какой executor держит нужные shuffle-блоки

// По умолчанию:
// spark.shuffle.manager = sort (SortShuffleManager)

// Замена на Celeborn:
// --conf spark.shuffle.manager=
//   org.apache.celeborn.client.spark.SparkShuffleManager

// Внутренние методы ShuffleManager:
// registerShuffle() → создаёт ShuffleHandle при начале shuffle
// getWriter()       → ShuffleWriter для map-задачи
// getReader()       → ShuffleReader для reduce-задачи
// unregisterShuffle() → очистка после завершения задания

// Три пути SortShuffleManager:
// 1. BypassMergeSort: нет агрегации И partition < 200
//    → spark.shuffle.sort.bypassMergeThreshold = 200
// 2. UnsafeShuffle: serializedSorting поддерживается кодеком
//    И нет агрегации И partition ≤ 16777216 (2^24)
// 3. SortShuffle: все остальные случаи (включая combiner)

// Результат всегда: один .data + один .index файл
// .index: массив кумулятивных смещений (8 байт × numPartitions+1)
// .data: конкатенация partition-блоков в порядке partition ID

// Поток данных MapOutputTracker:
// 1. Map task завершён → executor отправляет
//    MapStatus(blockManagerId, sizes[]) → Driver
// 2. Driver сохраняет в MapOutputTrackerMaster
// 3. Reduce task стартует → запрашивает у Driver:
//    GetMapOutputStatuses(shuffleId)
// 4. Driver отвечает сериализованным MapStatus[]
// 5. Executor распаковывает → знает, где читать каждый блок

// Мониторинг: spark.reducer.maxSizeInFlight контролирует
// параллельное чтение shuffle-блоков

// Включение push-based shuffle
spark.conf.set("spark.shuffle.push.enabled", "true")
spark.conf.set(
  "spark.shuffle.push.minShuffleSizeToWait", "500m"
)  // минимальный размер для активации push
spark.conf.set(
  "spark.shuffle.push.maxBlockSizeToPush", "1m"
)  // max размер блока для push

// Требует External Shuffle Service на каждом узле
// Метрики: remote merged fetches vs. remote fetches
// Spark UI → Stage → Task Metrics → Shuffle Read

# YARN: ESS встроен в NodeManager
spark.conf.set("spark.shuffle.service.enabled", "true")
spark.conf.set("spark.shuffle.service.port", "7337")

# Kubernetes: требует sidecar или DaemonSet
spark.conf.set("spark.shuffle.service.enabled", "true")
spark.conf.set(
  "spark.kubernetes.shuffle.namespace", "spark-shuffle"
)

# Без ESS на K8s — shuffle tracking:
spark.conf.set(
  "spark.dynamicAllocation.shuffleTracking.enabled", "true"
)

// ExternalSorter активируется в SortShuffleWriter:
// 1. insertAll() — добавляет записи с combiner (reduceByKey) или без
// 2. При нехватке памяти → maybeSpillCollection() → spill на диск
// 3. writePartitionedMapOutput() → merge-сортировка + единый .data файл

// Управление памятью spill:
spark.conf.set(
  "spark.shuffle.spill.compress", "true"
)  // сжатие spill-файлов
spark.conf.set(
  "spark.shuffle.sort.bypassMergeThreshold", "200"
)  // избежать sorter для < 200 partition

// Расчёт регионов памяти:
// executor.memory = 8g
// memory.fraction = 0.6 → 4.8g для UMM
// memory.storageFraction = 0.5 → 2.4g для storage (не вытесняется)
// Оставшиеся 40% (3.2g) — reserved для user code, overhead

spark.conf.set("spark.memory.fraction", "0.6")
spark.conf.set("spark.memory.storageFraction", "0.5")

// Мониторинг в Spark UI → Executors:
// Storage Memory: X.X GiB / Y.Y GiB
// Включить детали: /metrics/json endpoint

// Типы BlockId:
// RDDBlockId(rddId, partitionId)     → кеш RDD
// ShuffleBlockId(shuffleId, mapId, reduceId) → shuffle данные
// BroadcastBlockId(broadcastId)      → broadcast переменная
// TempShuffleBlockId(uuid)           → временный shuffle блок

// Чтение блока в порядке приоритета:
// 1. Локальная память (MemoryStore)
// 2. Локальный диск (DiskStore)
// 3. Remote BlockManager (по сети через BlockTransferService)
// 4. Пересчёт через lineage (для RDD-блоков)

// StorageLevel контролирует поведение MemoryStore:
df.persist(StorageLevel.MEMORY_ONLY)        // только heap, без disk
df.persist(StorageLevel.MEMORY_ONLY_SER)    // heap + сериализация
df.persist(StorageLevel.MEMORY_AND_DISK)    // heap → spill на диск
df.persist(StorageLevel.OFF_HEAP)           // off-heap (Tungsten)

// Мониторинг вытеснений:
// Spark UI → Storage → RDD Partitions:
// Cached In Memory: X, Cached On Disk: Y, Size in Memory: Z
// Если много Cached On Disk → memory pressure, увеличь executor.memory

// Настройка DiskStore:
spark.conf.set(
  "spark.local.dir", "/fast-ssd/spark-temp,/hdd/spark-temp"
)  // несколько директорий для striping

// StorageLevel с disk fallback:
df.persist(StorageLevel.MEMORY_AND_DISK)  // → DiskStore при вытеснении
df.persist(StorageLevel.DISK_ONLY)        // → только DiskStore

// Блоки на диске шифруются если:
spark.conf.set("spark.io.encryption.enabled", "true")
spark.conf.set("spark.io.encryption.keySizeBits", "128")
// Mониторинг: Spark UI → Executors → Disk Used

// Создание broadcast переменной
val lookup = spark.sparkContext.broadcast(
  Map("US" -> "United States", "DE" -> "Germany")
)  // → TorrentBroadcast: нарезается на блоки, хранится в driver BlockManager

// Использование на executor-ах
val result = rdd.map(row => lookup.value.getOrElse(row.country, "Unknown"))
// lookup.value → BlockManager читает локальный кеш или
//                тянет блоки у соседних executor-ов P2P

// Освобождение
lookup.unpersist()  // удалить с executor-ов, driver сохраняет
lookup.destroy()    // удалить полностью

// Просмотр всех стадий logical plan:
val df = spark.read.parquet("/sales")
  .filter("amount > 100")
  .groupBy("region").sum("amount")

val qe = df.queryExecution
println(qe.logical)         // unresolved: UnresolvedAttribute('amount')
println(qe.analyzed)        // resolved: amount#5 LongType
println(qe.optimizedPlan)   // optimized: predicate pushdown применён

// PySpark:
# df._jdf.queryExecution().logical()
# df._jdf.queryExecution().analyzed()
# df._jdf.queryExecution().optimizedPlan()

// Просмотр физического плана:
df.explain()         // краткий
df.explain(True)     // расширенный с codegen stage
df.explain('cost')   // с оценками стоимости

// Scala:
println(df.queryExecution.sparkPlan)    // до codegen
println(df.queryExecution.executedPlan) // после codegen

// Ключевые физические операторы:
// BroadcastHashJoin vs SortMergeJoin vs ShuffledHashJoin
// HashAggregate (с partial + final стадиями)
// SortExec (с spill support)
// Exchange (shuffle boundary)

// Просмотр всех фаз через DataFrame API:
import org.apache.spark.sql.execution.debug._

df.debugCodegen()  // распечатать сгенерированный код

// Через queryExecution:
val qe = df.queryExecution
println(qe.logical)          // unresolved logical plan
println(qe.analyzed)         // resolved logical plan
println(qe.optimizedPlan)    // optimized logical plan
println(qe.sparkPlan)        // physical plan (before codegen)
println(qe.executedPlan)     // final plan с CodeGenStage

// Структура optimizer rule (Scala):
class MyRule extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan =
    plan transform {
      // Pattern matching на узлах плана
      case Filter(Always True, child) => child
      case Project(cols, child)
        if cols == child.output => child
    }
}

// Регистрация кастомного правила:
spark.experimental.extraOptimizations =
  Seq(new MyRule)

// Или через SparkSessionExtensions:
.withExtensions(_.injectOptimizerRule(_ => new MyRule))

// Звёздочка (*) в EXPLAIN = CodeGenStage boundary:
// *(1) HashAggregate(keys=[dept], functions=[sum(salary)])
// +- Exchange hashpartitioning(dept, 200)
//    +- *(1) HashAggregate(keys=[dept], ...)
//       +- *(1) Filter (active = true)
//          +- *(1) ColumnarToRow
//             +- Scan parquet [...]

// Все *(1)-операторы скомпилированы в один метод!
// Exchange — граница, начинается новый CodeGenStage *(2)

// Отладка сгенерированного кода:
df.queryExecution.debug.codegen()

// Структура UnsafeRow для (id: INT, name: STRING, age: INT):
// Offset 0:  null bitmap (8 байт, 3 поля → 1 блок)
// Offset 8:  id value (8 байт, long-aligned)
// Offset 16: name offset+length (8 байт: [offset:32bit][len:32bit])
// Offset 24: age value (8 байт)
// Offset 32: name bytes (variable-length, выровнено до 8 байт)

// Использование в Spark:
// Все операции shuffle, sort, join работают с UnsafeRow
// Не нужна десериализация для сравнения ключей:
// UnsafeRowUtils.compare(row1, row2, keyFields)

// Размер можно смотреть в Spark UI: shuffle write bytes / num records

// Scala: неявные encoder-ы
import org.apache.spark.sql.Encoders
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder

case class User(id: Long, name: String, age: Int)

// Автоматически через implicit:
val ds = spark.createDataset(Seq(User(1, "Alice", 30)))

// Явно:
val enc = Encoders.product[User]
val exprEnc = enc.asInstanceOf[ExpressionEncoder[User]]

// Encoder кодогенерируется один раз и переиспользуется
// PySpark: PickleSerializer или ArrowSerializer (pandas UDF)
import pyspark.sql.functions as F

// Включение runtime фильтров:
spark.conf.set(
  "spark.sql.optimizer.runtimeFilter.enabled", "true"
)
spark.conf.set(
  "spark.sql.optimizer.runtimeFilter.creationSideThreshold",
  "10m"  // строить filter только для сторон < 10MB
)

// В EXPLAIN:
// Filter (isnotnull(id) AND might_contain(bf, xxhash64(id)))
// might_contain → bloom filter probe

// Bloom filter размер:
spark.conf.set(
  "spark.sql.optimizer.runtimeFilter.number.bloomFilter.bits",
  "8"  // bits per item
)

// Включение AQE (по умолчанию с Spark 3.2):
spark.conf.set("spark.sql.adaptive.enabled", "true")

// Три основных оптимизации AQE:
// 1. Coalesce shuffle partitions
spark.conf.set(
  "spark.sql.adaptive.coalescePartitions.enabled", "true"
)
spark.conf.set(
  "spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m"
)
// 2. Skew join detection & splitting
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
// 3. Runtime SortMergeJoin → BroadcastHashJoin
// Автоматически если размер < autoBroadcastJoinThreshold

spark.conf.set(
  "spark.sql.adaptive.coalescePartitions.enabled", "true"
)
spark.conf.set(
  "spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m"
)
spark.conf.set(
  "spark.sql.adaptive.coalescePartitions.minPartitionSize", "1m"
)
// minPartitionSize: нижняя граница — не объединять до мельче этого

// Результат: вместо 200 пустых partition-ов после join маленькой таблицы
// AQE может создать 3-5 реальных partition-ов
// Наблюдать в Spark UI → Stage Details → "Custom Metrics: numCoalescedPartitions"

spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set(
  "spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5"
)  // размер > 5x медианы = skew
spark.conf.set(
  "spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes",
  "256m"
)  // И > 256mb

// Наблюдение в Spark UI → SQL → DAG:
// SortMergeJoin (skew) → появляется при активации
// "number of skewed partitions" в Custom Metrics

// Ограничение: работает только с SortMergeJoin!
// BroadcastHashJoin skew не лечится AQE automaticaly

-- DPP автоматически применяется:
SELECT f.sale_amount, d.region_name
FROM fact_sales f
JOIN dim_region d ON f.region_id = d.id
WHERE d.country = 'DE';

-- Spark вставляет в план fact-таблицы:
-- Filter region_id IN (
--   DynamicPruning: SELECT id FROM dim_region WHERE country='DE'
-- )
-- → читает только partition-ы с немецкими region_id

-- Конфиги:
spark.conf.set(
  "spark.sql.optimizer.dynamicPartitionPruning.enabled", "true"
)

// MicroBatchExecution поток:
// 1. source.latestOffset() → получить новые offsets
// 2. source.getBatch(start, end) → DataFrame с новыми данными
// 3. IncrementalExecution.executedPlan → оптимизированный план
// 4. query.runBatch() → запуск как обычный Spark job
// 5. sink.addBatch(batchId, data) → запись результата
// 6. offsetLog.add(batchId, offsets) → checkpoint offsets
// 7. commitLog.add(batchId) → checkpoint commit

// Просмотр прогресса:
query.lastProgress  // метрики последнего micro-batch
query.recentProgress  // история последних N micro-batch

// HDFSBackedStateStore — по умолчанию:
// Хранит в checkpointLocation/state/<operatorId>/<partitionId>/
// Snapshot каждые N commit-ов, дельты между снапшотами
// Нагружает память executor (держит всё состояние в HashMap)

// RocksDB — для больших состояний:
spark.conf.set(
  "spark.sql.streaming.stateStore.providerClass",
  "org.apache.spark.sql.execution.streaming.state."
  + "RocksDBStateStoreProvider"
)
spark.conf.set(
  "spark.sql.streaming.stateStore.rocksdb.compactOnCommit", "false"
)
// Метрики: stateRows, stateMemory, loadedMapCacheHitCount

// IncrementalExecution добавляет в план:
// - StateStoreRestoreExec: читает предыдущее состояние из StateStore
// - HashAggregateExec: обрабатывает строки текущего batch
// - StateStoreSaveExec: сохраняет обновлённое состояние
// - WatermarkEvalExec: применяет eviction по watermark

// Просмотр плана streaming-запроса:
query.explain(extended=True)
// или через:
// spark.streams.active.head.explain()

// Структура checkpoint директории:
// checkpointLocation/
//   metadata          — id запроса и версия
//   offsets/          — WAL: offset лог
//     0, 1, 2 ...     — JSON с офсетами каждого batch
//   commits/          — commit лог
//     0, 1, 2 ...     — подтверждения завершённых batch
//   state/            — состояние StatefulOperator-ов
//   sources/          — source-specific состояние

// Два файла для batch N:
// offsets/N  существует → batch запущен
// commits/N  существует → batch завершён успешно

// Spark 4.0: transformWithState
class SessionAggregator
    extends StatefulProcessor[String, Event, SessionResult] {

  // Объявление state при инициализации
  var sessionCount: ValueState[Long] = _
  var sessionStart: ValueState[Long] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    sessionCount = getHandle.getValueState[Long]("count", Encoders.scalaLong)
    sessionStart = getHandle.getValueState[Long]("start", Encoders.scalaLong)
  }

  override def handleInputRows(key: String,
    rows: Iterator[Event], timerValues: TimerValues): Iterator[SessionResult] = {
    // Обновление state и генерация результатов
    ???
  }
}

// Конфигурация RocksDB backend:
spark.conf.set(
  "spark.sql.streaming.stateStore.providerClass",
  "org.apache.spark.sql.execution.streaming.state."
  + "RocksDBStateStoreProvider"
)

// Тюнинг RocksDB:
spark.conf.set(
  "spark.sql.streaming.stateStore.rocksdb.blockCacheSizeMB", "128"
)
spark.conf.set(
  "spark.sql.streaming.stateStore.rocksdb.writeBufferSizeBytes",
  str(2 * 1024 * 1024)  // 2MB write buffer
)
// Метрики: rocksdbNumSstFiles, rocksdbBytesWritten, rocksdbBlockCacheHit

// Включение Arrow для toPandas() и pandas UDF:
spark.conf.set(
  "spark.sql.execution.arrow.pyspark.enabled", "true"
)

# Без Arrow: каждая строка → pickle → Python O(n × row_size)
# С Arrow: RecordBatch → shared memory → zero-copy в Python

import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf("double")
def fast_udf(s: pd.Series) -> pd.Series:
    return s * 2.0  # выполняется над колонкой целиком (SIMD)

# Контроль размера batch:
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "10000")

import pyarrow as pa
from pyspark.sql.functions import pandas_udf
import pandas as pd

# mapInArrow: обработка данных как ArrowRecordBatch
def process_arrow(batches):
    for batch in batches:  # batch: pyarrow.RecordBatch
        arr = batch.column('amount').to_pylist()
        # нативная Arrow обработка без pandas overhead
        result = pa.array([x * 1.1 for x in arr])
        yield pa.record_batch(
            [result], schema=pa.schema([('amount', pa.float64())])
        )

df.mapInArrow(process_arrow, schema="amount double")

// Arrow Flight в контексте Spark Connect:
// Client (Python/Scala) → gRPC → Spark Connect Server
// Server → выполняет plan → возвращает через Arrow Flight Stream

// Схема взаимодействия:
// 1. client.execute(plan) → gRPC ExecutePlan RPC
// 2. Server: runQuery() → DataFrame
// 3. Server: serialize result → Arrow RecordBatches
// 4. Stream RecordBatches через gRPC → client
// 5. Client: pyarrow.RecordBatch → pandas DataFrame

// Конфигурация Flight endpoint:
spark.conf.set("spark.connect.grpc.binding.port", "15002")

# Подключение Spark Connect клиента:
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .remote("sc://spark-server:15002") \
    .getOrCreate()

# Тот же DataFrame API:
df = spark.read.parquet("s3://bucket/data")
result = df.filter("amount > 1000").groupBy("region").count()
result.show()

# Преимущества:
# - Client изолирован от cluster classpath конфликтов
# - Несколько клиентов на одном кластере
# - Клиент = любой язык с protobuf-поддержкой

// Регистрация расширений:
val spark = SparkSession.builder()
  .withExtensions { extensions =>
    // Кастомное optimizer rule:
    extensions.injectOptimizerRule(_ => new MyOptimizerRule)
    // Кастомная planner strategy:
    extensions.injectPlannerStrategy(_ => new MyPlannerStrategy)
    // Кастомный SQL parser:
    extensions.injectParser { (_, parser) =>
      new MyParser(parser)  // chain с дефолтным парсером
    }
    // Кастомный post-hoc analyzer rule:
    extensions.injectPostHocResolutionRule(_ => new MyAnalyzerRule)
  }
  .getOrCreate()

// Подключение Celeborn:
spark-submit \
  --conf spark.shuffle.manager=\
    org.apache.celeborn.client.spark.SparkShuffleManager \
  --conf spark.celeborn.master.endpoints=celeborn-master:9097 \
  --conf spark.celeborn.client.spark.push.data.timeout=120s \
  --conf spark.celeborn.client.spark.fetch.timeout=120s \
  my_app.py

// Преимущества vs ESS:
// - Горизонтально масштабируется (несколько Celeborn workers)
// - Replication для отказоустойчивости
// - Встроенный push-based shuffle
// - Совместим с K8s без DaemonSet

-- Принудительное использование SortMergeJoin:
SELECT /*+ MERGE(orders) */ *
FROM orders o JOIN products p ON o.product_id = p.id;

// В DataFrame API:
import org.apache.spark.sql.functions._
orders.join(products.hint("merge"), Seq("product_id"))

// Отключить broadcast для форсирования SMJ:
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

// SMJ можно ускорить Bucketing:
// Если обе таблицы bucketed by join key с одинаковым N buckets
// → shuffle пропускается (pre-sorted + pre-shuffled)

// Автоматический broadcast (таблица < threshold):
val result = big.join(small, "id")  // → BroadcastHashJoin

// Явный hint:
val result = big.join(broadcast(small), "id")

// SQL hint:
// SELECT /*+ BROADCAST(d) */ f.*, d.name
// FROM fact f JOIN dim d ON f.dim_id = d.id

// Настройка:
spark.conf.set(
  "spark.sql.autoBroadcastJoinThreshold", "50m"
)
// AQE может динамически upgrade SMJ → BHJ
// если после shuffle reduce-сторона < threshold

// Конфигурация таймаутов executor:
spark.conf.set("spark.executor.heartbeatInterval", "10s")
spark.conf.set("spark.network.timeout", "120s")
// network.timeout должен быть >> heartbeatInterval

// Executor thread pool:
spark.conf.set("spark.executor.cores", "4")  // 4 потока
// ThreadPoolExecutor с N = spark.executor.cores потоками
// Каждый поток = один TaskRunner = один task

// Graceful shutdown:
// Executor завершает текущие tasks, затем останавливается
// Принудительный kill: KillTask message от Driver

// Метрики spill в Spark UI → Stage → Task Metrics:
// Shuffle Spill (Memory): объём данных до компрессии
// Shuffle Spill (Disk): объём данных после компрессии на диске
// Spill (Memory) >> Spill (Disk) → высокая сжимаемость

// Предотвращение spill:
spark.conf.set("spark.executor.memory", "16g")  // больше памяти
spark.conf.set("spark.memory.fraction", "0.7")   // больше для execution

// Если spill неизбежен — ускорить:
spark.conf.set(
  "spark.shuffle.spill.compress", "true"
)
spark.conf.set("spark.io.compression.codec", "lz4")  // быстрая компрессия

// Включение Apache DataFusion Comet:
spark.conf.set(
  "spark.plugins",
  "org.apache.spark.CometPlugin"
)
spark.conf.set("spark.comet.enabled", "true")
spark.conf.set("spark.comet.exec.all.enabled", "true")

// Включение Gluten (Velox backend):
spark.conf.set(
  "spark.plugins",
  "io.glutenproject.GlutenPlugin"
)
spark.conf.set(
  "spark.gluten.sql.columnar.backend.lib", "velox"
)

// Наблюдение в EXPLAIN: NativeFilter, NativeScan вместо стандартных

// Включение Kryo:
spark.conf.set(
  "spark.serializer",
  "org.apache.spark.serializer.KryoSerializer"
)
// Регистрация классов для максимальной компактности:
spark.conf.set(
  "spark.kryo.classesToRegister",
  "com.example.User,com.example.Event"
)
// Или через KryoRegistrator:
class MyRegistrator extends KryoRegistrator {
  override def registerClasses(kryo: Kryo): Unit = {
    kryo.register(classOf[User])
    kryo.register(classOf[Event])
  }
}
spark.conf.set("spark.kryo.registrator", classOf[MyRegistrator].getName)

// Включение off-heap Tungsten:
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "8g")
// off-heap не входит в spark.executor.memory!
// Реальное потребление памяти executor-контейнера:
// = executor.memory + memoryOverhead + offHeap.size

// Кеш в off-heap:
df.persist(StorageLevel.OFF_HEAP)

// ВНИМАНИЕ: off-heap утечки не видны в GC-метриках
// Диагностика: native memory tracking
// -XX:NativeMemoryTracking=summary в extraJavaOptions

// Кастомный SparkListener (Scala):
import org.apache.spark.scheduler._

class CostTrackingListener extends SparkListener {
  override def onTaskEnd(event: SparkListenerTaskEnd): Unit = {
    val m = event.taskMetrics
    val shuffleWriteBytes = m.shuffleWriteMetrics.bytesWritten
    val gcTime = m.jvmGCTime
    // отправить в метрики систему (Prometheus, Datadog)
    MetricsClient.record("spark.task.shuffle_write", shuffleWriteBytes)
    MetricsClient.record("spark.task.gc_time_ms", gcTime)
  }
}

spark.sparkContext.addSparkListener(new CostTrackingListener())

// Подключение Uniffle к Spark:
spark-submit \
  --conf spark.shuffle.manager=\
    org.apache.uniffle.client.impl.grpc.ShuffleServerGrpcClient \
  --conf rss.coordinator.quorum=uniffle-coordinator:19999 \
  --conf rss.storage.type=MEMORY_LOCALFILE_HDFS \
  my_app.py

// Ключевые отличия от Celeborn:
// - Coordinator отдельный процесс (vs Celeborn master)
// - Поддержка HDFS как storage tier напрямую
// - Более гибкая система quota (по user/queue)

// Подключение Gluten (Velox backend):
spark-submit \
  --conf spark.plugins=io.glutenproject.GlutenPlugin \
  --conf spark.gluten.sql.columnar.backend.lib=velox \
  --conf spark.gluten.sql.columnar.forceShuffledHashJoin=true \
  --conf spark.sql.shuffle.partitions=200 \
  my_app.py

// Какие операторы нативны — видно в EXPLAIN:
// VeloxHashAggregateExec вместо HashAggregateExec
// VeloxBroadcastHashJoinExec вместо BroadcastHashJoinExec

// Метрики: veloxToNativeTime, veloxFromNativeTime — overhead конвертации

// Zero-copy toPandas() с Arrow:
spark.conf.set(
  "spark.sql.execution.arrow.pyspark.enabled", "true"
)

# Без Arrow: Row → pickle(bytes) → Python deserialize → object
# С Arrow: JVM allocates ArrowRecordBatch
#          → off-heap buffer
#          → Python pyarrow.RecordBatch wraps same memory
#          → pandas.DataFrame.from_pandas (column views)

# Измерение разницы:
import time
start = time.time()
pdf = df.toPandas()  # с Arrow: ~10-50x быстрее
print(f'toPandas: {time.time()-start:.2f}s')

// Структура DataSource V2 коннектора:
// 1. TableCatalog — регистрирует таблицы в Catalog
// 2. Table — возвращает Schema + Set[TableCapability]
// 3. SupportsRead → newScanBuilder()
// 4. ScanBuilder → build() → Scan
// 5. Scan → toBatch() → Batch
// 6. Batch → planInputPartitions() → InputPartition[]
// 7. PartitionReaderFactory → createReader(partition)

// Capabilities: BATCH_READ, BATCH_WRITE,
// OVERWRITE_BY_FILTER, TRUNCATE, V1_BATCH_WRITE

// Регистрация через SparkSessionExtensions или
// spark.sql.catalog.<name>=<CatalogImpl>

spark-submit \
  --conf spark.plugins=com.nvidia.spark.SQLPlugin \
  --conf spark.rapids.sql.enabled=true \
  --conf spark.rapids.memory.gpu.pooling.enabled=true \
  --conf spark.rapids.memory.gpu.pool.size=16G \
  --conf spark.executor.resource.gpu.amount=1 \
  --conf spark.task.resource.gpu.amount=0.25 \
  my_app.py

# Мониторинг GPU utilization:
# nvidia-smi dmon -s u  # GPU utilization per second
# Spark UI → SQL → "GPU" метрики в операторах