Monitoring Streaming Pipeline

Pipeline streaming TokoKita berjalan tanpa henti, dan penurunan kinerjanya pun senyap: consumer melambat sedikit demi sedikit, lag menumpuk, dan dashboard mulai menampilkan angka basi — tanpa satu pun proses yang gagal. Memantau pipeline streaming berarti mengukur jarak antara apa yang sudah diproduksi dan apa yang sudah diproses, lalu memasang ambang yang memberi tahu kita sebelum pengguna melihat datanya basi. Chapter ini menghitung consumer lag lewat simulasi nyata, membaca metrik lastProgress dari eksekusi Spark sungguhan, lalu menyusun ambang SLO dan strategi alert.

Consumer Lag: Jarak Terhadap Ujung Log #

Vokabulari topic, partition, dan committed offset dari Chapter Topics, Partitions, dan Offsets menjadi dasar metrik terpenting streaming. Consumer lag mengukur seberapa jauh sebuah consumer group tertinggal dari ujung log:

Istilah	Arti
head offset	offset pesan terbaru yang tertulis pada partisi
committed offset	offset terakhir yang di-commit oleh consumer group
lag per partisi	head offset − committed offset
total lag	jumlah lag seluruh partisi pada topic

Lag selalu ada dan itu normal — consumer hampir selalu tertinggal beberapa pesan dari producer. Yang dipantau adalah kecenderungannya: lag yang stabil kecil berarti konsumsi seimbang dengan produksi, sedangkan lag yang tumbuh linear berarti kapasitas konsumsi di bawah laju produksi dan antrean akan terus memanjang sampai retention menghapus pesan terlama yang belum terbaca.

Simulasi: Lag yang Tumbuh Lalu Pulih #

Simulasi berikut memodelkan topic pesanan-tokokita dengan 3 partisi: producer menambah 4 pesanan per partisi per detik, consumer group laporan-harian mula-mula hanya sanggup memproses 2 per partisi per detik, lalu kapasitasnya dinaikkan menjadi 9 pada detik ke-13:

python
"""Simulasi consumer lag: head offset vs committed offset per partition."""
LAJU = 4              # pesanan baru per partisi per detik (producer)
AMBANG = 50           # ambang notifikasi total lag


def tick(head, committed, kapasitas):
    """Satu detik: produksi dulu, lalu konsumsi per partisi."""
    for p in range(3):
        head[p] += LAJU
        bisa = min(kapasitas, head[p] - committed[p])
        committed[p] += bisa


head = [0, 0, 0]
committed = [0, 0, 0]
cetak = {0, 4, 8, 12, 16, 20, 24}
sudah_alert = False

print("topik pesanan-tokokita, 3 partisi, consumer group laporan-harian")
print("fase 1 (t<=12): kapasitas konsumsi 2 pesanan/partisi/detik")
for t in range(1, 25):
    if t == 13:
        print("fase 2 (t>=13): kapasitas naik menjadi 9 pesanan/partisi/detik")
    tick(head, committed, 2 if t <= 12 else 9)
    lag = [h - c for h, c in zip(head, committed)]
    total = sum(lag)
    if t in cetak:
        print(f"t={t:>2}  head={head} committed={committed} "
              f"lag={lag} total={total}")
    if total > AMBANG and not sudah_alert:
        sudah_alert = True
        print(f"    ALERT di t={t}: total lag {total} melewati ambang {AMBANG}")
    if total <= AMBANG and sudah_alert:
        sudah_alert = False
        print(f"    pulih di t={t}: total lag {total} kembali di bawah ambang")

print()
print("kesimpulan: lag tumbuh selama kapasitas konsumsi di bawah laju "
      "produksi, lalu menyusut setelah kapasitas dinaikkan")

Output yang Dihasilkan:

text
topik pesanan-tokokita, 3 partisi, consumer group laporan-harian
fase 1 (t<=12): kapasitas konsumsi 2 pesanan/partisi/detik
t= 4  head=[16, 16, 16] committed=[8, 8, 8] lag=[8, 8, 8] total=24
t= 8  head=[32, 32, 32] committed=[16, 16, 16] lag=[16, 16, 16] total=48
    ALERT di t=9: total lag 54 melewati ambang 50
t=12  head=[48, 48, 48] committed=[24, 24, 24] lag=[24, 24, 24] total=72
fase 2 (t>=13): kapasitas naik menjadi 9 pesanan/partisi/detik
    pulih di t=14: total lag 42 kembali di bawah ambang
t=16  head=[64, 64, 64] committed=[60, 60, 60] lag=[4, 4, 4] total=12
t=20  head=[80, 80, 80] committed=[80, 80, 80] lag=[0, 0, 0] total=0
t=24  head=[96, 96, 96] committed=[96, 96, 96] lag=[0, 0, 0] total=0

kesimpulan: lag tumbuh selama kapasitas konsumsi di bawah laju produksi, lalu menyusut setelah kapasitas dinaikkan

Mari kita telaah baris-baris kode di atas:

Baris 6-11 (def tick): satu detik dipecah dua — produksi menambah head, lalu konsumsi sebanyak min(kapasitas, lag); min menjaga committed offset tidak pernah melampaui head.
Baris 22-24 (tick(head, committed, 2 if t <= 12 else 9)): kapasitas konsumsi naik dari 2 ke 9 mulai detik ke-13, meniru penambahan consumer atau vertikal scaling.
Baris 30-35 (if total > AMBANG ...): alert hanya berbunyi saat kondisi melanggar, sekali, lalu mencatat pemulihan — bukan berbunyi terus setiap detik.

Fase 1 memperlihatkan pertumbuhan linear: produksi 4 dikurangi konsumsi 2 menyisakan 2 pesanan per partisi per detik, sehingga total lag bertambah 6 tiap detik — dari 24 di detik ke-4 hingga menembus ambang 50 pada detik ke-9. Fase 2 membalik arah: kapasitas 9 melampaui produksi 4, lag menyusut 5 per partisi per detik, pulih di bawah ambang pada detik ke-14 dan mencapai nol pada detik ke-17. Di Kafka sungguhan, angka-angka ini terbaca lewat kafka-consumer-groups.sh --describe yang menampilkan kolom LAG per partisi, terdokumentasi di dokumentasi Apache Kafka .

lastProgress: Denyut Internal Spark #

Sisi engine juga mengukur dirinya sendiri. Setiap StreamingQuery Spark mengekspos lastProgress — kamus metrik batch terakhir — dan status. Skrip berikut menjalankan kueri pendek dengan sink noop, lalu mencetak sebagian isi lastProgress:

python
"""Membaca metrik StreamingQuery: lastProgress dan status (run nyata)."""
import json
import time

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .master("local[1]")
         .config("spark.ui.enabled", "false")
         .config("spark.sql.shuffle.partitions", "2")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

t0 = time.time()

daftar = F.array(F.lit("Rok Tenun"), F.lit("Kain Songket"),
                 F.lit("Tas Rotan"), F.lit("Blouse Katun"))
pesanan = (spark.readStream.format("rate")
           .option("rowsPerSecond", 5)
           .option("numPartitions", 1)
           .load()
           .withColumn("produk", F.element_at(daftar, (F.col("value") % 4 + 1).cast("int"))))

q = (pesanan.writeStream
     .format("noop")
     .trigger(processingTime="3 seconds")
     .start())
q.awaitTermination(13)
q.stop()

p = q.lastProgress
metrik = {
    "batchId": p["batchId"],
    "numInputRows": p["numInputRows"],
    "inputRowsPerSecond": round(p["inputRowsPerSecond"], 2),
    "processedRowsPerSecond": round(p["processedRowsPerSecond"], 2),
    "durationMs": p["durationMs"],
    "sumber": p["sources"][0]["description"],
    "sink": p["sink"]["description"],
}
print(json.dumps(metrik, indent=2, default=str))
print("status:", q.status["message"])
print(f"durasi script: {time.time() - t0:.1f} detik")
spark.stop()

Output yang Dihasilkan:

text
{
  "batchId": 4,
  "numInputRows": 15,
  "inputRowsPerSecond": 5.0,
  "processedRowsPerSecond": 157.9,
  "durationMs": {
    "addBatch": 30,
    "commitOffsets": 25,
    "getBatch": 0,
    "latestOffset": 0,
    "queryPlanning": 14,
    "triggerExecution": 95,
    "walCommit": 23
  },
  "sumber": "RateStreamV2[rowsPerSecond=5, rampUpTimeSeconds=0, numPartitions=1",
  "sink": "NoopTable"
}
status: Stopped
durasi script: 15.4 detik

Empat angka di situ bercerita. numInputRows 15 pada trigger 3 detik cocok dengan inputRowsPerSecond 5.0 — kueri seimbang dengan sumbernya. processedRowsPerSecond 157.9 jauh di atas laju masuk karena sink noop memang tidak menulis apa pun; pada sink sungguhan, angka ini yang dibandingkan dengan laju masuk. durationMs membedah 95 milidetik eksekusi trigger menjadi tahapannya — addBatch mengolah data, walCommit menulis checkpoint offset, commitOffsets menandai batch selesai, queryPlanning menyusun rencana — sehingga pelambatan bisa dilokalisasi ke tahap mana. status tercetak Stopped karena dibaca setelah q.stop(). Metrik agregasi state — termasuk numRowsDroppedByWatermark dari Chapter Watermarks dan Late Events — berada di kunci stateOperators dari objek progress yang sama. Sistem metrik Spark beserta cara mengekspornya ke luar terdokumentasi di halaman Monitoring Spark .

Ambang SLO dan Strategi Alert #

Monitoring tanpa ambang hanyalah tumpukan grafik. SLO yang lazim untuk pipeline streaming mengikat angka-angka di atas ke batas yang bisa ditindaklanjuti:

Metrik	Contoh SLO	Contoh pemicu alert
total consumer lag	di bawah 10.000 pesan	di atas 10.000 selama 5 menit
lag dalam satuan waktu	di bawah 60 detik	di atas 120 detik selama 5 menit
processedRowsPerSecond	minimal setara laju masuk	di bawah inputRowsPerSecond selama 10 menit
numRowsDroppedByWatermark	mendekati nol	lonjakan tajam di luar pola harian

Dua kebiasaan membuat alert bisa dipercaya. Pertama, picu pada pelanggaran yang berkelanjutan, bukan lonjakan sesaat — burst traffic singkat bukan insiden, dan alert yang sering salah membuat tim mengabaikannya. Kedua, nyatakan lag dalam satuan waktu, bukan hanya jumlah pesan, dengan membagi lag terhadap laju produksi.

Di luar alat bawaan, pipeline produksi umumnya mengekspor metrik ke Prometheus dan menampilkannya di Grafana — consumer lag dari Kafka, metrik batch dari Spark — sehingga ambang di tabel di atas menjadi aturan alert yang berjalan otomatis. Seluruh hasil pipeline — window agregat, hasil join, laporan yang selalu segar — akhirnya bermuara di satu tempat yang bisa di-query cepat oleh analis. Section berikutnya, Database Analitik dan Query Engine, membedah tempat bermuara itu: penyimpanan kolom, eksekusi vectorized, hingga cara memilih query engine.