Block Locality: как Spark scheduler использует data-locality при планировании

from pyspark.sql import SparkSession
from pyspark import StorageLevel

spark = SparkSession.builder.getOrCreate()

# Первое чтение: NODE_LOCAL (данные с HDFS)
df = spark.read.parquet("hdfs://cluster/data/events/")

# Кешируем в памяти Executor'ов.
# После этого повторные обращения к df дадут PROCESS_LOCAL
# для партиций, которые уже закешированы в конкретных Executor'ах.
df.cache()
df.count()  # триггерит материализацию кеша

# Теперь этот запрос выполнится с PROCESS_LOCAL для закешированных партиций
result = df.groupBy("event_type").count()
result.show()

#!/bin/bash
# /etc/hadoop/topology.sh
# Скрипт определения топологии стойки по IP-адресу.
# HDFS вызывает этот скрипт при регистрации каждого DataNode.

IP=$1  # IP адрес DataNode передаётся как аргумент

case $IP in
    # Rack 1: серверы 192.168.1.0/24
    192.168.1.*)  echo "/rack1" ;;
    # Rack 2: серверы 192.168.2.0/24
    192.168.2.*)  echo "/rack2" ;;
    # Rack 3: серверы 192.168.3.0/24
    192.168.3.*)  echo "/rack3" ;;
    # По умолчанию: все неизвестные в rack0
    *)            echo "/default-rack" ;;
esac

<!-- core-site.xml -->
<property>
  <name>net.topology.script.file.name</name>
  <value>/etc/hadoop/topology.sh</value>
</property>

<!--
  Количество IP-адресов, которое кешируется в топологии.
  Для кластеров > 100 нод увеличивайте это значение во избежание
  постоянных вызовов скрипта при heartbeat'ах DataNode.
-->
<property>
  <name>net.topology.script.number.of.cached.filenames</name>
  <value>100</value>
</property>

INFO TaskSetManager: Starting task 42.0 in stage 3.0 (TID 185, server4,
  executor 7, partition 42, RACK_LOCAL, 5102 bytes)

WARN TaskSetManager: Level for task 42 has been downgraded from RACK_LOCAL
  to ANY because it waited 3000ms without finding a RACK_LOCAL executor.

INFO TaskSetManager: Starting task 42.0 in stage 3.0 (TID 186, server7,
  executor 12, partition 42, ANY, 5102 bytes)

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("locality-tuning") \

    # spark.locality.wait - базовое время ожидания для всех уровней.
    # Применяется если не установлены специфичные параметры ниже.
    # Дефолт: 3 сек (3000мс). Значение "0" = немедленный переход к ANY.
    .config("spark.locality.wait", "3s") \

    # spark.locality.wait.process - время ожидания для PROCESS_LOCAL.
    # Имеет смысл только при работе с кешированными DataFrame.
    # Для джобов, читающих "холодные" данные с HDFS, установите в 0 —
    # PROCESS_LOCAL недостижим при первом чтении с диска.
    .config("spark.locality.wait.process", "5s") \

    # spark.locality.wait.node - время ожидания для NODE_LOCAL.
    # Наиболее важный параметр для HDFS-кластеров.
    # При загруженных нодах увеличьте до 5-10 секунд.
    # При коротких задачах (< 1 сек каждая) уменьшите —
    # ожидание 3 секунды больше самой задачи неэффективно.
    .config("spark.locality.wait.node", "10s") \

    # spark.locality.wait.rack - время ожидания для RACK_LOCAL.
    # Обычно достаточно дефолта. Снижайте если rack-local
    # сеть перегружена и ANY не намного хуже.
    .config("spark.locality.wait.rack", "3s") \

    .getOrCreate()

Spark Executor (JVM)
    ↓
HDFS Client в JVM
    ↓ TCP-соединение (localhost:50010)
DataNode процесс (JVM)
    ↓ read() syscall
Файл на диске: /data/hdfs/current/blk_001234

<!-- hdfs-site.xml на всех DataNode и клиентских узлах (Spark) -->

<!--
  Включаем Short-Circuit Reads для HDFS клиентов.
  true: HDFS Client будет пытаться читать блоки напрямую если DataNode
  на том же хосте, иначе автоматически откатывается к TCP.
-->
<property>
  <name>dfs.client.read.shortcircuit</name>
  <value>true</value>
</property>

<!--
  Путь к Unix Domain Socket для коммуникации между HDFS Client и DataNode.
  DataNode создаёт этот сокет при старте.
  Путь должен существовать и быть доступен HDFS-пользователю.
-->
<property>
  <name>dfs.domain.socket.path</name>
  <value>/var/lib/hadoop-hdfs/dn_socket</value>
</property>

<!--
  Разрешить DataNode чтение файлов блоков от имени других пользователей.
  Необходимо если Spark работает под другим UNIX пользователем, чем DataNode.
-->
<property>
  <name>dfs.block.local-path-access.user</name>
  <value>spark,yarn,mapred</value>
</property>

<!--
  Выполнять checksum проверку при short-circuit чтении.
  false: пропустить проверку для максимальной скорости (риск corrupt data)
  true: выполнять проверку (рекомендуется для production)
-->
<property>
  <name>dfs.client.read.shortcircuit.skip.checksum</name>
  <value>false</value>
</property>

spark = SparkSession.builder \
    .config("spark.hadoop.dfs.client.read.shortcircuit", "true") \
    .config("spark.hadoop.dfs.domain.socket.path", "/var/lib/hadoop-hdfs/dn_socket") \
    .getOrCreate()

# Метрики DataNode через JMX (порт 9864 для HDFS 3.x)
curl -s "http://datanode1:9864/jmx?qry=Hadoop:service=DataNode,name=DataNodeActivity" \
  | python3 -m json.tool \
  | grep -E "BlocksRead|ShortCircuit"

# Ожидаемый вывод при активном Short-Circuit:
# "BlocksRead" : 15234,
# "ShortCircuitLocalReadsRead" : 14891,  # ~97% от BlocksRead
# "ShortCircuitLocalBytesRead" : 1234567890

# Просмотр распределения данных по DataNode
hdfs dfsadmin -report

# Пример вывода с явным дисбалансом:
# DataNode1: DFS Used%: 78%  ← Hotspot!
# DataNode2: DFS Used%: 71%
# DataNode3: DFS Used%: 12%
# DataNode4: DFS Used%: 15%

# Запуск балансировщика HDFS
hdfs balancer -threshold 10
# threshold 10: балансировка завершена если разница < 10%

spark = SparkSession.builder \
    # Включить спекулятивное выполнение
    .config("spark.speculation", "true") \

    # Задача считается «отстающей», если выполняется в multiplier
    # раз дольше медианы всех задач в Stage.
    # Дефолт: 1.5 (на 50% медленнее медианы = кандидат)
    .config("spark.speculation.multiplier", "1.5") \

    # Запускать спекуляцию только если завершено не менее fraction задач Stage.
    # При 0.75: ждём завершения 75% задач прежде чем решить «задача медленная».
    .config("spark.speculation.quantile", "0.75") \

    .getOrCreate()

# Первое выполнение: Stage 0 (NODE_LOCAL) + Stage 1 (ANY после Shuffle)
silver_df = (
    spark.read.parquet("hdfs://cluster/bronze/events/")
    .filter(df.event_type == "purchase")
    .groupBy("user_id")
    .agg({"amount": "sum", "event_id": "count"})
)

# Кешируем - данные хранятся в памяти Executor'ов
silver_df.cache()
silver_df.count()  # материализуем кеш

# Последующие операции с silver_df получат PROCESS_LOCAL:
# данные уже в heap Executor'ов, никакого I/O нет
result1 = silver_df.join(users_df, "user_id")
result2 = silver_df.join(products_df, "user_id")

Locality Level Distribution:
  NODE_LOCAL: 847 задач (92%)
  RACK_LOCAL:  62 задачи  (7%)
  ANY:         11 задач   (1%)
Среднее Scheduler Delay: 45ms

Locality Level Distribution:
  NODE_LOCAL:  123 задачи (13%)
  RACK_LOCAL:  234 задачи (25%)
  ANY:         623 задачи (66%)  ← проблема!
Среднее Scheduler Delay: 3200ms  ← долгое ожидание

import requests
from collections import Counter


def get_locality_distribution(app_id: str, stage_id: int,
                               history_server_url: str) -> dict:
    """
    Получает распределение уровней локальности для заданного Stage.
    Полезно для автоматизированного аудита после завершения джобы.
    """
    url = (f"{history_server_url}/api/v1/applications/{app_id}"
           f"/stages/{stage_id}/0/taskList")
    tasks = requests.get(url).json()

    locality_counts = Counter(task["taskLocality"] for task in tasks)
    total = sum(locality_counts.values())

    distribution = {
        level: {
            "count": count,
            "percentage": round(count / total * 100, 1)
        }
        for level, count in locality_counts.items()
    }

    # Задачи с аномально высоким Scheduler Delay
    high_delay_threshold_ms = 5000
    slow_tasks = [
        {
            "taskId": t["taskId"],
            "locality": t["taskLocality"],
            "schedulerDelayMs": t["taskMetrics"]["schedulerDelay"],
            "durationMs": t["duration"],
        }
        for t in tasks
        if t.get("taskMetrics", {}).get("schedulerDelay", 0) > high_delay_threshold_ms
    ]

    return {
        "stageId": stage_id,
        "totalTasks": total,
        "localityDistribution": distribution,
        "slowTasksCount": len(slow_tasks),
        "slowTasks": slow_tasks[:5],
    }

# Запуск Spark с детальным логированием планировщика
spark-submit \
  --conf "spark.driver.extraJavaOptions=-Dlog4j.logger.org.apache.spark.scheduler=DEBUG" \
  your_job.py

# Задача получила NODE_LOCAL:
INFO  TaskSetManager: Starting task 5.0 in stage 0.0 (TID 15,
  datanode3.example.com, executor 3, partition 5, NODE_LOCAL, 5242 bytes)

# Снижение требований к локальности:
WARN  TaskSetManager: No NODE_LOCAL tasks are launchable, waiting 3000ms
INFO  TaskSetManager: Level for task 4 has been downgraded from NODE_LOCAL
  to RACK_LOCAL because it waited 3000ms without finding a suitable executor

# Спекулятивный запуск задачи:
INFO  TaskSetManager: Marking task 12 in stage 1.0 (on server2) as
  speculative because it ran more than 15.0s
INFO  TaskSetManager: Starting task 12.1 in stage 1.0 (TID 156,
  server5.example.com, executor 8, partition 12, ANY, 5102 bytes)

# Пороги деградации для автоматического мониторинга
LOCALITY_THRESHOLDS = {
    "node_local_min_pct": 70,        # минимум 70% задач NODE_LOCAL
    "any_max_pct": 15,               # не более 15% задач ANY
    "fetch_wait_max_ratio": 0.3,     # Fetch Wait Time < 30% от Duration
    "scheduler_delay_max_ms": 3000,  # Scheduler Delay < 3 сек
}

# production_spark_config.py
# Производственная конфигурация SparkSession с учётом Data Locality

from pyspark.sql import SparkSession


def create_spark_hdfs_optimized(
    app_name: str,
    nameservice: str = "mycluster",
    task_duration_seconds: float = 30.0,
    enable_short_circuit: bool = True,
    enable_speculation: bool = False,
) -> SparkSession:
    """
    Создаёт SparkSession с настройками для максимальной HDFS локальности.

    task_duration_seconds: ожидаемое среднее время выполнения одной задачи.
    Используется для расчёта оптимального spark.locality.wait.

    Правило выбора locality.wait:
    - Если задача < 5 сек:  wait = 0 (ожидание бессмысленно)
    - Если задача 5-30 сек: wait = 3-5 сек
    - Если задача > 30 сек: wait = 5-15 сек
    """
    if task_duration_seconds < 5:
        locality_wait = "0"
        locality_wait_node = "0"
    elif task_duration_seconds < 30:
        locality_wait = "3s"
        locality_wait_node = "5s"
    else:
        locality_wait = "5s"
        locality_wait_node = "15s"

    builder = SparkSession.builder \
        .appName(app_name) \
        .config("spark.hadoop.fs.defaultFS", f"hdfs://{nameservice}") \
        # PROCESS_LOCAL при первом чтении HDFS недостижим → ставим 0
        .config("spark.locality.wait.process", "0") \
        .config("spark.locality.wait.node", locality_wait_node) \
        .config("spark.locality.wait.rack", locality_wait) \
        .config("spark.locality.wait", locality_wait) \
        .config("spark.task.maxFailures", "4")

    if enable_short_circuit:
        builder = builder \
            .config("spark.hadoop.dfs.client.read.shortcircuit", "true") \
            .config("spark.hadoop.dfs.domain.socket.path",
                    "/var/lib/hadoop-hdfs/dn_socket") \
            .config("spark.hadoop.dfs.client.read.shortcircuit.skip.checksum",
                    "false")

    if enable_speculation:
        builder = builder \
            .config("spark.speculation", "true") \
            .config("spark.speculation.multiplier", "2.0") \
            .config("spark.speculation.quantile", "0.9")
    else:
        builder = builder.config("spark.speculation", "false")

    return builder.getOrCreate()


# Для batch ETL с долгими задачами (агрегации, joins на десятках GB)
spark_batch = create_spark_hdfs_optimized(
    app_name="daily-etl",
    task_duration_seconds=60.0,
    enable_short_circuit=True,
)

# Для Structured Streaming (критично: wait=0 чтобы не задерживать микробатчи)
spark_streaming = create_spark_hdfs_optimized(
    app_name="kafka-to-hdfs-stream",
    task_duration_seconds=1.0,
    enable_short_circuit=True,
    enable_speculation=False,  # Speculation и Streaming несовместимы
)

# lab_locality_wait_comparison.py
# Сравниваем влияние spark.locality.wait на производительность и
# распределение уровней локальности.

import time
from dataclasses import dataclass
from pyspark.sql import SparkSession
from pyspark.sql import functions as F


@dataclass
class BenchmarkResult:
    config_name: str
    locality_wait: str
    duration_seconds: float


def run_benchmark(spark: SparkSession, data_path: str,
                  config_name: str, locality_wait: str) -> BenchmarkResult:
    """
    Запускает стандартный аналитический запрос с заданным locality.wait.
    Измеряет время выполнения.
    """
    spark.conf.set("spark.locality.wait", locality_wait)
    spark.conf.set("spark.locality.wait.node", locality_wait)

    print(f"\nКонфигурация: {config_name}, locality.wait = {locality_wait}")
    start_time = time.time()

    events = spark.read.parquet(f"{data_path}/events/")
    users = spark.read.parquet(f"{data_path}/users/")

    result = (
        events
        .filter(F.col("event_type").isin("purchase", "checkout"))
        .join(users.select("user_id", "country", "segment"), "user_id", "left")
        .groupBy("country", "segment",
                 F.date_trunc("day", "event_time").alias("day"))
        .agg(
            F.count("*").alias("event_count"),
            F.sum("amount").alias("total_revenue"),
            F.countDistinct("session_id").alias("unique_sessions"),
        )
        .orderBy("day", "country")
    )

    row_count = result.count()
    duration = time.time() - start_time

    print(f"Результат: {row_count} строк за {duration:.1f} секунд")
    return BenchmarkResult(config_name, locality_wait, duration)


def generate_benchmark_data(spark: SparkSession, output_path: str,
                             n_events: int = 5_000_000) -> None:
    """Генерирует тестовый датасет для бенчмарка на HDFS."""
    print(f"Генерируем {n_events:,} событий → {output_path}")

    events = spark.range(n_events).select(
        F.col("id").alias("event_id"),
        (F.rand() * 100_000).cast("long").alias("user_id"),
        F.array(
            F.lit("purchase"), F.lit("view"), F.lit("checkout"), F.lit("search")
        ).getItem((F.rand() * 4).cast("int")).alias("event_type"),
        (F.rand() * 1000).alias("amount"),
        F.expr("uuid()").alias("session_id"),
        F.current_timestamp().alias("event_time"),
    )

    # Создаём много небольших файлов для теста локальности
    events.repartition(200).write.mode("overwrite").parquet(f"{output_path}/events/")

    users = spark.range(100_000).select(
        F.col("id").alias("user_id"),
        F.array(F.lit("RU"), F.lit("DE"), F.lit("US")).getItem(
            (F.rand() * 3).cast("int")
        ).alias("country"),
        F.array(F.lit("premium"), F.lit("standard"), F.lit("free")).getItem(
            (F.rand() * 3).cast("int")
        ).alias("segment"),
    )
    users.coalesce(10).write.mode("overwrite").parquet(f"{output_path}/users/")
    print("Данные записаны")


def print_comparison(results: list[BenchmarkResult]) -> None:
    print("\n" + "=" * 60)
    print("СРАВНЕНИЕ КОНФИГУРАЦИЙ LOCALITY.WAIT")
    print("=" * 60)
    print(f"{'Конфигурация':<30} {'Wait':>6} {'Время':>10}")
    print("-" * 60)
    for r in results:
        print(f"{r.config_name:<30} {r.locality_wait:>6} {r.duration_seconds:>9.1f}s")

    fastest = min(results, key=lambda r: r.duration_seconds)
    print(f"\nБыстрейшая конфигурация: {fastest.config_name}")


if __name__ == "__main__":
    spark = SparkSession.builder \
        .master("yarn") \
        .appName("locality-wait-benchmark") \
        .config("spark.hadoop.fs.defaultFS", "hdfs://mycluster") \
        .config("spark.executor.instances", "20") \
        .config("spark.executor.cores", "4") \
        .config("spark.executor.memory", "8g") \
        .config("spark.hadoop.dfs.client.read.shortcircuit", "true") \
        .config("spark.hadoop.dfs.domain.socket.path",
                "/var/lib/hadoop-hdfs/dn_socket") \
        .getOrCreate()

    DATA_PATH = "hdfs://mycluster/lab/locality-benchmark/"
    generate_benchmark_data(spark, DATA_PATH, n_events=5_000_000)

    configs = [
        ("Без ожидания (wait=0)", "0"),
        ("Базовый (wait=3s)", "3s"),
        ("Агрессивный (wait=10s)", "10s"),
        ("Максимальный (wait=30s)", "30s"),
    ]

    results = []
    for name, wait in configs:
        result = run_benchmark(spark, DATA_PATH, name, wait)
        results.append(result)
        time.sleep(5)

    print_comparison(results)
    spark.stop()

# 1. Распределение данных по DataNode (ищем Hotspot)
hdfs dfsadmin -report | grep -A5 "DataNode"

# 2. Балансировка данных (при дисбалансе > 15%)
hdfs balancer -threshold 10 &
tail -f /var/log/hadoop/hdfs/hadoop-hdfs-balancer-*.log

# 3. Block locations конкретного файла
hdfs fsck /data/events/part-00001.parquet -files -blocks -locations

# 4. Проверить Short-Circuit метрики DataNode
curl -s "http://datanode1.example.com:9864/jmx" \
  | python3 -c "
import sys, json
data = json.load(sys.stdin)
for bean in data['beans']:
    if 'DataNodeActivity' in bean.get('name', ''):
        for k in ['BlocksRead', 'ShortCircuitLocalReadsRead']:
            print(f'{k}: {bean.get(k, 0):,}')
"

# 5. Topology: rack ID каждого DataNode
hdfs dfsadmin -printTopology

# 6. Работающие Spark приложения
yarn application -list -appStates RUNNING