Apache Uniffle и сравнение решений

Что такое Uniffle

Apache Uniffle — remote shuffle service, изначально разработанный в Tencent. Uniffle graduated как Apache Top-Level Project (TLP) в феврале 2025 года. Последняя стабильная версия — 0.9.1.

Uniffle решает те же фундаментальные проблемы shuffle, что и Celeborn (L02), но с архитектурными отличиями: другая модель координации, 3-tier storage (Memory + Local + HDFS), и отличающаяся стратегия организации данных.

Архитектура

Uniffle имеет более простую серверную архитектуру: Coordinator для координации и Shuffle Server для хранения данных. Клиентская часть встроена в Spark Driver и Executor через client library.

Coordinator cluster

Coordinator — координатор кластера Uniffle. В отличие от Celeborn Master (Raft consensus), Coordinator работает через heartbeat-based модель:

• Shuffle Servers регулярно отправляют heartbeat с текущей нагрузкой (CPU, memory, disk usage)
• Coordinator назначает подходящие Shuffle Servers для нового shuffle, балансируя нагрузку на основе heartbeat-данных
• HA обеспечивается кластером Coordinator-ов

Конфигурация Coordinator:

# coordinator.conf — конфигурация Uniffle Coordinator
rss.coordinator.quorum=coord-1:19999,coord-2:19999,coord-3:19999
rss.coordinator.server.heartbeat.timeout=30000

# Стратегия назначения Shuffle Servers
rss.coordinator.assignment.strategy=PARTITION_BALANCE
rss.coordinator.app.expired=60000

# Мониторинг
rss.coordinator.metrics.reporter.class=\
  org.apache.uniffle.common.metrics.PrometheusMetricsReporter

Shuffle Servers

Shuffle Server — основной рабочий компонент Uniffle:

• Принимает shuffle-данные от map tasks
• Merge данные для одной partition
• Записывает index файл (метаданные блоков) и data файл (содержимое)
• Поддерживает 3-tier storage: Memory -> Local Disk -> HDFS

Конфигурация Shuffle Server:

# server.conf — конфигурация Uniffle Shuffle Server
rss.coordinator.quorum=coord-1:19999,coord-2:19999,coord-3:19999
rss.storage.type=MEMORY_LOCALFILE_HDFS
rss.storage.basePath=/data/rss/shuffle

# Memory management — watermark thresholds для spill
rss.server.buffer.capacity=16g
rss.server.memory.shuffle.highWaterMark.percentage=75.0
rss.server.memory.shuffle.lowWaterMark.percentage=25.0

# Flush configuration
rss.server.flush.threadPool.size=10
rss.server.heartbeat.interval=10000

# HDFS fallback storage
rss.server.hdfs.base.path=hdfs:///rss/shuffle-data

Каждый Shuffle Server отчитывается Coordinator-у о своём состоянии через heartbeat, что позволяет Coordinator балансировать нагрузку и исключать unhealthy servers.

Data Flow

Uniffle реализует 5-шаговый data flow:

1. Query Coordinator: Spark Driver запрашивает у Coordinator назначение Shuffle Servers для нового shuffle. Coordinator выбирает servers на основе heartbeat-данных (нагрузка, доступное место)
2. Buffer and push: Map tasks буферизуют KV-данные локально. Когда буфер заполняется, данные flush-атся в очередь и отправляются на назначенные Shuffle Servers
3. Cache and write: Shuffle Server кэширует данные в memory, затем записывает на local disk и/или HDFS (в зависимости от storage type). Создаются index файл (метаданные каждого блока) и data файл (содержимое)
4. Organize by partition: Data файл содержит блоки для конкретной partition. Index файл хранит смещения и размеры каждого блока — это позволяет эффективно читать отдельные блоки
5. Read phase: Reduce tasks запрашивают данные у Shuffle Server, remote storage (HDFS), или обоих — в зависимости от конфигурации

Storage Backends

Уникальная особенность Uniffle — поддержка нескольких уровней хранения (tiered storage):

3-tier storage (рекомендация)

MEMORY_LOCALFILE_HDFS — рекомендованная конфигурация для production:

1. Данные сначала буферизуются в memory Shuffle Server — максимально быстрый доступ
2. При заполнении memory данные flush-атся на local disk (SSD/HDD) — быстрый random access
3. Данные реплицируются в HDFS — distributed durability, выживание при потере Shuffle Server

Этот подход даёт лучший баланс производительности и надёжности: hot данные в memory, warm на local disk, cold в HDFS.

Конфигурация Spark

spark-submit

spark-submit \
  --conf spark.shuffle.manager=org.apache.spark.shuffle.RssShuffleManager \
  --conf spark.rss.coordinator.quorum=coord-1:19999,coord-2:19999 \
  --conf spark.rss.storage.type=MEMORY_LOCALFILE_HDFS \
  --conf spark.rss.writer.buffer.size=3m \
  --conf spark.rss.client.read.buffer.size=14m \
  --conf spark.rss.client.send.size.limit=16m \
  --conf spark.rss.writer.buffer.spill.size=128m \
  --conf spark.rss.data.replica.write=1 \
  --conf spark.rss.data.replica.read=1 \
  --jars rss-client-spark3-shaded.jar \
  your_app.py

Ключевые параметры

Адаптивное подключение и Dynamic Allocation

# DelegationRssShuffleManager — автоматический fallback
spark.shuffle.manager=org.apache.spark.shuffle.DelegationRssShuffleManager
spark.rss.access.id=my_spark_app
spark.rss.access.timeout.ms=10000

# Dynamic Allocation с Uniffle (Spark 3.5+)
spark.shuffle.service.enabled=false
spark.dynamicAllocation.enabled=true
spark.shuffle.sort.io.plugin.class=\
  org.apache.spark.shuffle.RssShuffleDataIo

Client Quorum (Fault Tolerance)

# Quorum-based replication — блоки отправляются на N серверов
spark.rss.data.replica.write=2
spark.rss.data.replica.read=2
spark.rss.data.replica.skip.enabled=true

# Reader может получить данные с любого из N серверов,
# обеспечивая resilience при потере отдельного Shuffle Server

Поддержка версий

Uniffle поддерживает Spark 2.3.x, 2.4.x, 3.0.x—3.5.x. Также поддерживается Hadoop MapReduce 2/3.

Production adoption

Сравнительный анализ

Celeborn vs Uniffle vs Built-in ESS

Decision Framework

Когда использовать Built-in ESS:

• Small-to-medium кластеры (до 50 executor-ов)
• Moderate shuffle volumes (гигабайты, не терабайты)
• Нет бюджета на отдельный shuffle-кластер
• Workload допускает occasional recomputation

Когда использовать Apache Celeborn:

• Large-scale кластеры с shuffle-heavy workloads
• Требуется поддержка Spark 4.0/4.1
• Mixed workload: Spark + Flink на одном shuffle-кластере
• Интеграция с Apache Gluten для full native stack (М15/L03)
• Нужен mature community и широкий набор production adopters

Когда использовать Apache Uniffle:

• Hadoop-centric environment с существующей HDFS инфраструктурой
• 3-tier storage критически важна (memory -> local -> HDFS)
• Spark 2.3—3.5 (не требуется Spark 4.x)
• Tencent ecosystem / community

Cross-references

Этот модуль (М16) связан с другими модулями курса:

• М04/L01 (Оптимизация shuffle) — основы shuffle, Exchange в physical plan, spark.sql.shuffle.partitions. M16 продолжает тему, разбирая ограничения встроенного shuffle
• М14/L06 (Альтернативные движки и расширения Spark) — awareness-level обзор ESS и push-based shuffle. М16 даёт полный deep-dive
• М15/L03 (Apache Gluten) — Celeborn integration с Gluten для full native stack (Velox/ClickHouse + remote shuffle)

Итоги модуля

Модуль М16 «External Shuffle Service» охватил:

1. L01: Фундаментальные ограничения встроенного shuffle (lifecycle coupling, disk I/O, recomputation), ESS и push-based shuffle, зачем нужны remote shuffle сервисы
2. L02: Apache Celeborn — Master/Worker/LifecycleManager архитектура, push-based flow, Revive fault tolerance, конфигурация, Spark 2.4—4.1 + Flink
3. L03: Apache Uniffle — Coordinator/Shuffle Server, 3-tier storage, конфигурация, и сравнительный анализ Celeborn vs Uniffle vs ESS

Ключевой вывод: remote shuffle services — critical infrastructure для large-scale Spark deployments. Выбор между Celeborn и Uniffle зависит от вашего stack (Spark version, Flink, Gluten), существующей инфраструктуры (HDFS), и масштаба. Для новых проектов на Spark 4.x с Flink и Gluten — Celeborn. Для Hadoop-centric сред с 3-tier storage требованиями — Uniffle.