Блок 7 - Kafka, Streaming и мониторинг

Kafka Cluster
├── Broker 1 (Kafka Server)        ← физический сервер Kafka
│   ├── Topic "events"             ← логическая очередь
│   │   ├── Partition 0 [LEADER]   ← физический лог
│   │   │   ├── Segment 1.log      ← файл (~1 GB)
│   │   │   ├── Segment 2.log      ← закрытый
│   │   │   └── Segment 3.log      ← активный (текущая запись)
│   │   ├── Partition 1 [FOLLOWER] ← реплика (на другом Broker)
│   │   └── Partition 2 [LEADER]
│   └── Topic "orders"
├── Broker 2
└── Broker 3

Topic "events": [Partition 0] [Partition 1] [Partition 2]

Group "analytics":
  Consumer A → Partition 0 (offset 100)
  Consumer B → Partition 1, 2 (offset 50, 75)

Group "reporting":
  Consumer X → Partition 0, 1, 2 (читает независимо от Group "analytics")

# С ключом: partition = hash(key) % num_partitions
producer.send("events", key=b"user_123", value=event_json)
# hash("user_123") % 3 = 1 → всегда Partition 1
# Гарантия: все события user_123 → одна партиция → сохранён порядок

# Без ключа: round-robin по партициям
producer.send("events", value=event_json)
# Message 1 → Partition 0
# Message 2 → Partition 1
# Message 3 → Partition 2
# Message 4 → Partition 0 (снова)

# По времени (приоритет: ms > minutes > hours)
log.retention.hours=168          # 7 дней (default)
log.retention.ms=604800000       # эквивалентно, но приоритетнее

# По размеру
log.retention.bytes=1073741824   # 1 GB на партицию (-1 = без лимита)

До compaction (один из сегментов):
offset 1: key=user1, value={"age": 25}
offset 2: key=user2, value={"name": "Bob"}
offset 5: key=user1, value={"age": 26}   ← обновление
offset 8: key=user1, value={"age": 27}   ← ещё обновление

После compaction:
offset 2: key=user2, value={"name": "Bob"}  (единственное значение)
offset 8: key=user1, value={"age": 27}      (только последнее)

# foreachBatch - самый гибкий вариант
def write_batch(batch_df, batch_id):
    batch_df.write.parquet(f"s3a://bucket/out/batch={batch_id}")
    batch_df.write.jdbc(url, table, "append")  # два sink'а!

stream.writeStream.foreachBatch(write_batch).start()

# Backfill pattern: обработать всё и завершить
query = stream.writeStream \
    .trigger(once=True) \
    .format("parquet") \
    .option("checkpointLocation", "s3a://bucket/ckpt/") \
    .start()
query.awaitTermination()

query = stream.writeStream \
    .format("iceberg") \
    .option("checkpointLocation", "s3a://bucket/checkpoints/my_stream/") \
    .start()
# Checkpoint хранит: последний обработанный offset для каждой partition/topic
# При рестарте: читаем с offset'а после last committed

conda create -n myenv python=3.10 && conda activate myenv
pip install pandas numpy scikit-learn
conda pack -o myenv.tar.gz

spark-submit \
    --archives myenv.tar.gz#environment \
    --conf spark.executorEnv.PYSPARK_PYTHON=./environment/bin/python \
    job.py

python -m venv myenv && source myenv/bin/activate
pip install -r requirements.txt
venv-pack -o myenv.tar.gz

spark-submit \
    --archives myenv.tar.gz#venv \
    --conf spark.executorEnv.PYSPARK_PYTHON=./venv/bin/python \
    job.py

from sparkmeasure import StageMetrics

stage_metrics = StageMetrics(spark)
stage_metrics.begin()

# Ваш код
df.groupBy("key").agg(F.sum("amount")).collect()

stage_metrics.end()
stage_metrics.print_report()
# → executorRunTime: 4512 ms
# → shuffleReadBytes: 1.2 GB
# → jvmGCTime: 312 ms
# → peakExecutionMemory: 2.8 GB
# → memoryBytesSpilled: 0

spark.metrics.namespace = spark.app.id
# Пример: spark.application_1680000000000_0001

spark.metrics.namespace=${spark.app.name}
# daily_etl_sales → всегда spark.daily_etl_sales.executor.*

# metrics.properties
*.sink.graphite.class=org.apache.spark.metrics.sink.GraphiteSink
*.sink.graphite.host=influxdb.internal
*.sink.graphite.port=2003
*.sink.graphite.period=10
*.sink.graphite.unit=seconds

LocalExecutor: Scheduler → [fork] → Task Process (та же машина)
CeleryExecutor: Scheduler → [Redis Queue] → Worker Node 1, 2, N
KubernetesExecutor: Scheduler → K8s API → Pod per Task (создаётся/удаляется)

# ❌ Неверно: чтение больших данных на Airflow-машине
def process_data():
    import pandas as pd
    df = pd.read_csv("s3://bucket/100gb_file.csv")  # 100 GB на Airflow-сервере!
    result = df.groupby("key").sum()

task = PythonOperator(task_id="process", python_callable=process_data)

# ❌ Неверно: Spark работает прямо на Airflow-машине
task = BashOperator(
    task_id="run_spark",
    bash_command="python /opt/jobs/etl.py"  # Нет кластера, нет параллелизма!
)

# ✓ Вариант 1: KubernetesPodOperator - изолированный контейнер
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

task = KubernetesPodOperator(
    task_id="etl",
    image="myregistry/spark-job:v1.2.3",
    cmds=["python", "/app/etl.py"],
    arguments=["--date", "{{ ds }}"],
)

# ✓ Вариант 2: SparkSubmitOperator - submit на кластер
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

task = SparkSubmitOperator(
    task_id="etl",
    application="s3a://bucket/jobs/etl.py",
    conf={"spark.executor.memory": "4g"},
)

Client → spark-submit → YARN ResourceManager
  → NodeManager → YARN Container
      ├── Spark Application Master
      └── Spark Driver (здесь, в кластере)
         → Запрашивает Executor контейнеры у ResourceManager

# Список контейнеров приложения
yarn logs -applicationId application_1680000000000_0001

# Логи конкретного контейнера
yarn logs -applicationId application_1680000000000_0001 \
          -containerId container_e01_1680000000000_0001_01_000002

# Признаки OOM в stderr Executor:
# java.lang.OutOfMemoryError: Java heap space
# GC overhead limit exceeded
# Container killed by YARN for exceeding memory limits

# Просмотр в Spark UI:
# Executors tab → stdout/stderr links

Spark Driver / Executor
    ↓ (GraphiteSink - metrics.properties)
InfluxDB (Graphite input plugin, порт 2003)
    ↓
Grafana Dashboard
    ↓ (алерты)
PagerDuty / Slack

# Минимальная конфигурация для production мониторинга
spark.conf.set("spark.metrics.namespace", "${spark.app.name}")
spark.conf.set("spark.eventLog.enabled", "true")
spark.conf.set("spark.eventLog.dir", "s3a://bucket/spark-logs/")
# metrics.properties настраивается отдельно с GraphiteSink

Запрос: SELECT ... FROM FACT JOIN DIM ON FACT.date_id = DIM.date_id WHERE DIM.year = 2023

Без DPP:                          С DPP:
1. Read ALL FACT (100 GB)         1. Read DIM WHERE year = 2023
2. Read DIM WHERE year = 2023        → date_id ∈ {1, 2, 5, 7, ...}
3. Join + filter                  2. Read FACT WHERE date_id IN (...) [20 GB!]
                                  3. Join (уже отфильтрованные данные)

spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true")  # default ON

# Чтение из Kafka
stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "events,orders") \
    .option("startingOffsets", "earliest")  # или "latest"
    .option("maxOffsetsPerTrigger", 100000)  # ограничить размер batch
    .load()

# stream_df.schema: key, value, topic, partition, offset, timestamp, ...
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, LongType

schema = StructType().add("user_id", LongType()).add("event", StringType())

parsed = stream_df.select(
    from_json(col("value").cast("string"), schema).alias("data"),
    col("partition"),
    col("offset")
).select("data.*", "partition", "offset")

# Запись обратно в Kafka
result = parsed.groupBy("event").count()

result.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("topic", "event_counts") \
    .option("checkpointLocation", "s3a://bucket/ckpt/kafka_out/") \
    .outputMode("update") \
    .start()

# Проверить что pushdown работает
df.filter("year = 2024").select("user_id").explain()
# В плане:
# PushedFilters: [EqualTo(year, 2024)]
# ReadSchema: struct<user_id:bigint>   ← только нужная колонка

Cores на Executor: 5
Executors на узел: (16 - 1) / 5 = 3  ← 1 core оставить на ОС/YARN
RAM на Executor: (64 - 4) / 3 ≈ 20 GB  ← 4 GB на ОС
memoryOverhead: 2 GB
executor.memory: 18 GB