Skew Detection: как увидеть skew в Task Duration и shuffle read

Summary Metrics for Completed Tasks
────────────────────────────────────────────────────────────────────────
Metric              | Min    | 25th %ile | Median | 75th %ile | Max
────────────────────────────────────────────────────────────────────────
Duration            | 1.2 s  | 3.4 s     | 4.1 s  | 5.0 s     | 42 min
GC Time             | 0.1 s  | 0.3 s     | 0.4 s  | 0.5 s     | 8 min
Shuffle Read Size   | 1.2 MB | 3.8 MB    | 4.5 MB | 5.2 MB    | 3.8 GB
Shuffle Read Records| 12K    | 38K       | 44K    | 51K       | 38.2M
Input Size          | 0 B    | 0 B       | 0 B    | 0 B       | 0 B
Spill (Memory)      | 0 B    | 0 B       | 0 B    | 0 B       | 12.4 GB
Spill (Disk)        | 0 B    | 0 B       | 0 B    | 0 B       | 2.1 GB
────────────────────────────────────────────────────────────────────────

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, count, desc, lit, when, rand, concat_ws,
    percentile_approx, sum as spark_sum, round as spark_round,
    spark_partition_id, countDistinct
)
import pyspark.sql.functions as F

spark = SparkSession.builder \
    .appName("skew-detection") \
    .config("spark.sql.shuffle.partitions", "50") \
    .getOrCreate()

# Создаём датасет с искусственным skew:
# user_id = "bot_001" встречается в 40% строк
# user_id = "guest" встречается в 20% строк
# Остальные 40% - нормальные уникальные пользователи
from pyspark.sql.types import *

n_rows = 2_000_000

# Генерируем строки с перекосом
clickstream = spark.range(n_rows).select(
    col("id").alias("event_id"),
    when(rand() < 0.40, lit("bot_001"))         # 40% - один бот
    .when(rand() < 0.60, lit("guest"))           # 20% (0.60 - 0.40) - гость
    .when(rand() < 0.70, lit("unknown"))         # 10% - unknown
    .otherwise(concat_ws("_", lit("user"), (col("id") % 50000).cast("string")))
    .alias("user_id"),
    (rand() * 100).alias("page_id"),
    F.current_timestamp().alias("event_ts"),
)

clickstream.cache()
clickstream.count()  # материализуем

# Топ-20 значений ключа по количеству строк
key_distribution = clickstream \
    .groupBy("user_id") \
    .count() \
    .withColumn("pct_of_total", spark_round(col("count") / n_rows * 100, 2)) \
    .orderBy(desc("count"))

key_distribution.show(20, truncate=False)

+---------+--------+------------+
|  user_id|   count|pct_of_total|
+---------+--------+------------+
|  bot_001| 800234|       40.01|
|    guest| 399876|       19.99|
|  unknown| 199843|        9.99|
|  user_0 |     47|        0.00|
|  user_1 |     42|        0.00|
|  user_2 |     38|        0.00|
|  ...    |    ...|         ...|
+---------+--------+------------+

# Агрегат: сколько уникальных ключей и как распределены строки
key_stats = clickstream \
    .groupBy("user_id") \
    .count() \
    .agg(
        F.count("user_id").alias("distinct_keys"),
        F.min("count").alias("min_rows_per_key"),
        F.max("count").alias("max_rows_per_key"),
        F.avg("count").alias("avg_rows_per_key"),
        percentile_approx("count", 0.5).alias("median_rows"),
        percentile_approx("count", 0.95).alias("p95_rows"),
        percentile_approx("count", 0.99).alias("p99_rows"),
    )

key_stats.show()

+------------+----------------+----------------+----------------+-----------+---------+---------+
|distinct_keys|min_rows_per_key|max_rows_per_key|avg_rows_per_key|median_rows|p95_rows |p99_rows |
+------------+----------------+----------------+----------------+-----------+---------+---------+
|       50003|               1|          800234|            39.9|         40|       47|      200|
+------------+----------------+----------------+----------------+-----------+---------+---------+

from pyspark.sql.functions import isnull

null_analysis = clickstream.agg(
    count("*").alias("total_rows"),
    count(when(isnull("user_id"), True)).alias("null_user_id"),
    count(when(col("user_id") == "unknown", True)).alias("unknown_user_id"),
    count(when(col("user_id") == "guest", True)).alias("guest_user_id"),
).withColumn(
    "null_pct",
    spark_round(col("null_user_id") / col("total_rows") * 100, 2)
).withColumn(
    "unknown_pct",
    spark_round(col("unknown_user_id") / col("total_rows") * 100, 2)
)

null_analysis.show()

+----------+------------+---------------+-------------+--------+-----------+
|total_rows|null_user_id|unknown_user_id|guest_user_id|null_pct|unknown_pct|
+----------+------------+---------------+-------------+--------+-----------+
|   2000000|           0|         199843|       399876|     0.0|       9.99|
+----------+------------+---------------+-------------+--------+-----------+

# Приблизительный объём shuffle по ключу
shuffle_estimate = clickstream \
    .groupBy("user_id") \
    .agg(
        count("*").alias("row_count"),
        # Оцениваем ~100 байт на строку (event_id + user_id + page_id + ts)
        (count("*") * 100).alias("estimated_shuffle_bytes"),
    ) \
    .orderBy(desc("row_count")) \
    .limit(10)

shuffle_estimate.withColumn(
    "estimated_shuffle_mb",
    spark_round(col("estimated_shuffle_bytes") / 1024 / 1024, 1)
).select("user_id", "row_count", "estimated_shuffle_mb").show()

+---------+--------+--------------------+
|  user_id|row_count|estimated_shuffle_mb|
+---------+--------+--------------------+
|  bot_001| 800234|                76.3|
|    guest| 399876|                38.1|
|  unknown| 199843|                19.1|
|  user_0 |      47|                 0.0|
+---------+--------+--------------------+

# Таблица профилей пользователей
user_profiles = spark.range(50000).select(
    concat_ws("_", lit("user"), col("id")).alias("user_id"),
    (rand() * 10).cast("int").alias("age_group"),
    F.element_at(F.array(lit("RU"), lit("US"), lit("DE")),
                 ((col("id") % 3) + 1).cast("int")).alias("country")
)

# Добавляем "горячих" пользователей
hot_users = spark.createDataFrame([
    ("bot_001", 99, "XX"),
    ("guest", 0, "XX"),
    ("unknown", 0, "XX"),
], ["user_id", "age_group", "country"])

all_profiles = user_profiles.union(hot_users)

# JOIN: clickstream × profiles
# Это МЕДЛЕННЫЙ join - будет skew
spark.conf.set("spark.sql.adaptive.enabled", "false")  # отключаем AQE для наглядности

result = clickstream.join(
    all_profiles,
    on="user_id",
    how="left"
).groupBy("country").count()

result.show()

# Смотрим распределение ПОСЛЕ shuffle (partition distribution post-join)
result_with_pid = clickstream.join(
    all_profiles.hint("skew", "user_id"),
    on="user_id",
    how="left"
).select(spark_partition_id().alias("pid"), "user_id") \

partition_sizes = result_with_pid \
    .groupBy("pid") \
    .agg(
        count("*").alias("rows_in_partition"),
        countDistinct("user_id").alias("distinct_keys"),
    ) \
    .orderBy(desc("rows_in_partition"))

# Топ-5 самых больших партиций
partition_sizes.show(10)

+----+------------------+-------------+
| pid| rows_in_partition|distinct_keys|
+----+------------------+-------------+
|  7 |           800234|            1|  ← bot_001
| 31 |           399876|            1|  ← guest
| 42 |           199843|            1|  ← unknown
|  0 |             1245|           31|
|  1 |             1198|           30|
|  2 |             1267|           32|
|  3 |             1231|           31|
+----+------------------+-------------+

# Skew в groupBy
skewed_agg = clickstream \
    .groupBy("user_id") \
    .agg(
        count("*").alias("event_count"),
        F.max("event_ts").alias("last_event"),
    )

# Измеряем размеры партиций в результате агрегации
skewed_agg.select(spark_partition_id().alias("pid"), "user_id") \
    .groupBy("pid") \
    .agg(count("*").alias("keys_in_partition")) \
    .orderBy(desc("keys_in_partition")) \
    .show(5)

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank

# ОПАСНО: partitionBy("user_id") при skewed user_id
window_spec = Window.partitionBy("user_id").orderBy("event_ts")

# Весь bot_001 - 800K строк - попадёт в одну партицию
# Executor должен держать в памяти весь отсортированный массив для user_id
result_with_rank = clickstream.withColumn(
    "session_rank",
    row_number().over(window_spec)
)
# Этот код выполнится, но будет очень медленным и, вероятно, вызовет spill

def assess_skew(df, key_col: str, sample_fraction: float = 1.0) -> None:
    """
    Профилирует распределение ключа и оценивает severity skew.
    При большом датасете можно использовать sample_fraction < 1.0.
    """
    if sample_fraction < 1.0:
        sample_df = df.sample(fraction=sample_fraction, seed=42)
    else:
        sample_df = df

    key_counts = sample_df.groupBy(key_col).count()

    stats = key_counts.agg(
        count(key_col).alias("distinct_keys"),
        F.min("count").alias("min_count"),
        F.max("count").alias("max_count"),
        F.avg("count").alias("avg_count"),
        percentile_approx("count", 0.5).alias("p50"),
        percentile_approx("count", 0.9).alias("p90"),
        percentile_approx("count", 0.99).alias("p99"),
    ).collect()[0]

    max_to_median = stats["max_count"] / max(stats["p50"], 1)
    p99_to_p50 = stats["p99"] / max(stats["p50"], 1)

    print(f"\n=== Skew Assessment for column: {key_col} ===")
    print(f"  Distinct keys:     {stats['distinct_keys']:,}")
    print(f"  Min rows per key:  {stats['min_count']:,}")
    print(f"  Median rows:       {stats['p50']:,}")
    print(f"  P99 rows:          {stats['p99']:,}")
    print(f"  Max rows per key:  {stats['max_count']:,}")
    print(f"  Max/Median ratio:  {max_to_median:.0f}x")
    print(f"  P99/Median ratio:  {p99_to_p50:.0f}x")

    if max_to_median > 100:
        severity = "CRITICAL - немедленное вмешательство"
    elif max_to_median > 20:
        severity = "HIGH - требует исправления"
    elif max_to_median > 5:
        severity = "MEDIUM - стоит исследовать"
    else:
        severity = "LOW - приемлемо"

    print(f"\n  Severity: {severity}")

    # Топ hot keys
    print(f"\n  Top-10 hot keys:")
    top_keys = key_counts.orderBy(desc("count")).limit(10).collect()
    total = sample_df.count() if sample_fraction < 1.0 else df.count()
    for row in top_keys:
        pct = row["count"] / total * 100
        bar = "█" * int(pct / 2)
        print(f"    {str(row[key_col]):20s} {row['count']:>10,} rows ({pct:5.1f}%) {bar}")


# Применяем
assess_skew(clickstream, "user_id")

=== Skew Assessment for column: user_id ===
  Distinct keys:     50,003
  Min rows per key:  1
  Median rows:       40
  P99 rows:          200
  Max rows per key:  800,234
  Max/Median ratio:  20006x

  Severity: CRITICAL - немедленное вмешательство

  Top-10 hot keys:
    bot_001              800,234 rows (40.0%) ████████████████████
    guest                399,876 rows (20.0%) ██████████
    unknown              199,843 rows (10.0%) █████
    user_12345               47 rows ( 0.0%)
    user_28901               46 rows ( 0.0%)

# Для 1 TB датасета - сэмплируем 1%, результат статистически надёжен
sample_fraction = 0.01
assess_skew(huge_df, "join_key", sample_fraction=sample_fraction)

# Или approximate distinct count через HLL
from pyspark.sql.functions import approx_count_distinct

quick_stats = clickstream.agg(
    count("*").alias("total"),
    approx_count_distinct("user_id", rsd=0.05).alias("approx_distinct_users"),
).show()

# В spark-defaults.conf или при создании SparkSession
spark = SparkSession.builder \
    .config("spark.eventLog.enabled", "true") \
    .config("spark.eventLog.dir", "s3a://bucket/spark-logs/") \
    .getOrCreate()

from pyspark.sql.functions import count, max as spark_max, min as spark_min

def log_partition_stats(df, label: str):
    """Логирует распределение партиций для мониторинга."""
    stats = df.select(spark_partition_id().alias("pid")) \
        .groupBy("pid").count() \
        .agg(
            spark_min("count").alias("min_rows"),
            spark_max("count").alias("max_rows"),
            F.avg("count").alias("avg_rows"),
            count("pid").alias("num_partitions"),
        ).collect()[0]

    skew_ratio = stats["max_rows"] / max(stats["avg_rows"], 1)
    print(f"[{label}] partitions={stats['num_partitions']}, "
          f"min={stats['min_rows']}, avg={stats['avg_rows']:.0f}, "
          f"max={stats['max_rows']}, skew_ratio={skew_ratio:.1f}x")
    if skew_ratio > 10:
        print(f"  ⚠ WARNING: high skew detected (ratio={skew_ratio:.0f}x)")


# Используем перед тяжёлыми операциями
log_partition_stats(clickstream.repartition(50, "user_id"), "before_join")

Stage 5: "SortMergeJoin [user_id]"
  Duration: 1h 38m
  Tasks: 200/200 completed

  Task Duration:
    Min: 0.8s  |  25%: 2.1s  |  Median: 3.4s  |  75%: 4.2s  |  Max: 1h 37m

  Shuffle Read Size:
    Min: 0.5MB  |  25%: 1.8MB  |  Median: 2.3MB  |  75%: 2.9MB  |  Max: 6.2GB

  Shuffle Read Records:
    Min: 5K  |  25%: 18K  |  Median: 23K  |  75%: 29K  |  Max: 61.8M

  Spill (Memory): Max = 28.4GB
  Spill (Disk):   Max = 4.7GB

  Associated query: events.join(users, on="user_id", how="left")

# 1. Подтверждаем наличие skew по квантилям
max_duration_sec = 97 * 60  # 1h 37m в секундах
median_duration_sec = 3.4
skew_ratio_time = max_duration_sec / median_duration_sec
print(f"Task Duration Skew: {skew_ratio_time:.0f}x")  # → 1706x

max_shuffle_bytes = 6.2 * 1024  # MB
median_shuffle_bytes = 2.3
skew_ratio_shuffle = max_shuffle_bytes / median_shuffle_bytes
print(f"Shuffle Read Skew: {skew_ratio_shuffle:.0f}x")  # → 2762x

max_records = 61_800_000
median_records = 23_000
skew_ratio_records = max_records / median_records
print(f"Records Skew: {skew_ratio_records:.0f}x")  # → 2687x

# 2. Оцениваем объём spill
print(f"\nSpill to disk: 4.7 GB - критично для производительности")
print(f"Spill to memory: 28.4 GB - executor работает за пределами памяти")

# 3. Делаем вывод
print("\n=== Заключение ===")
print("Диагноз: Data Skew в SortMergeJoin [user_id]")
print(f"  Severity: CRITICAL (time ratio: {skew_ratio_time:.0f}x)")
print(f"  Проблемный ключ: 1 значение user_id содержит 61.8M строк")
print(f"    vs медиана 23K строк - это 2687x превышение нормы")
print(f"  Следствие: Executor вышел за пределы памяти:")
print(f"    Spill(Disk) = 4.7 GB, Spill(Memory) = 28.4 GB")
print(f"  Рекомендация: проверить распределение user_id,")
print(f"    изолировать NULL/sentinel значения,")
print(f"    применить AQE skewJoin или salting")

# Джоб работает за 2 часа - "нормально", никто не исследует
result = huge_events.join(profiles, on="user_id")
result.write.parquet(...)

# На самом деле: медиана задач - 3 секунды, один task - 1.5 часа
# При росте данных в 2x время вырастет не в 2x, а в 10-20x
# Skew усиливается нелинейно с ростом данных

# "Перераспределяем" 1000 партиций - не помогает при hot keys
df.repartition(1000).join(other, on="hot_key")

# Если hash("bot_001") % 1000 = 42, все 800K строк всё равно пойдут
# в partition 42. Количество партиций не решает проблему hot keys.

# НЕПРАВИЛЬНО: сразу запускаем join на терабайтном датасете
result = tb_scale_events.join(profiles, on="user_id")  # упадёт через час

# ПРАВИЛЬНО: сначала профилируем ключ на 1% sample
assess_skew(tb_scale_events.sample(0.01), "user_id")
# Потом принимаем архитектурное решение: AQE / broadcast / salting

# "Джоб завершился, всё ок"
# На самом деле: Spill(Disk) = 50 GB, время в 5x дольше оптимального
# Spill - это молчаливый убийца производительности