Menerapkan marka air untuk mengontrol ambang batas pemrosesan data

Halaman ini menjelaskan konsep watermark dan memberikan rekomendasi tentang penggunaan watermark dalam operasi streaming stateful yang umum.

Kueri streaming mengakumulasi data status dari waktu ke waktu. Marka air secara otomatis menghapus data status lama untuk mencegah kesalahan memori dan peningkatan latensi pemrosesan.

Apa itu marka air?

Selama pemrosesan, Streaming Terstruktur menjaga status di seluruh batch mikro. Kueri streaming menggunakan state untuk memperbarui hasil secara bertahap, alih-alih menghitung ulang semuanya setelah setiap mikro-batch. Watermark menentukan ambang batas saat kueri berhenti memproses entitas state.

Contoh umum entitas negara meliputi:

• Agregasi dalam rentang waktu.
• Kunci unik dalam penggabungan antara dua aliran data.

Untuk mendeklarasikan watermark pada DataFrame streaming, tentukan kolom stempel waktu dan batas keterlambatan. Saat data baru tiba, pengelola status melacak stempel waktu terbaru pada bidang yang ditentukan dan hanya memproses catatan data yang masih berada dalam batas keterlambatan.

Kueri selalu memproses data yang masuk dalam nilai ambang. Kueri mungkin masih memproses data yang masuk di luar ambang batas, namun hal ini tidak dijamin.

Contoh berikut menerapkan ambang batas marka air 10 menit ke jumlah berjendela:

Python

from pyspark.sql.functions import window

(df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window("event_time", "5 minutes"),
    "id")
  .count()
)

Scala

import org.apache.spark.sql.functions.window

df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window($"event_time", "5 minutes"),
    $"id")
  .count()

Dalam contoh ini:

• Kolom event_time digunakan untuk menentukan marka air 10 menit dan jendela bergulir 5 menit.
• Hitungan dikumpulkan untuk setiap id yang diamati pada setiap periode 5 menit yang tidak tumpang tindih.
• Informasi status dipertahankan untuk setiap hitungan hingga akhir jendela lebih lama 10 menit dari yang terbaru diamati event_time.

Bagaimana tanda air memengaruhi waktu dan kapasitas pemrosesan?

Mode keluaran menentukan kapan kueri dengan watermark menulis data ke tujuan. Watermark sangat penting untuk pengendalian throughput dalam pemrosesan aliran stateful karena mengurangi jumlah total informasi state di memori. Tidak semua mode keluaran didukung untuk seluruh operasi berbasis status. Lihat Tanda air dan mode output untuk agregasi berjendela.

Pemilihan durasi tanda air memiliki kompromi:

• Watermark yang lebih singkat menurunkan latensi kueri karena kueri menyimpan lebih sedikit informasi status dan menulis hasil setelah setiap durasi watermark berakhir. Namun, watermark pendek memiliki toleransi yang rendah terhadap data yang datang terlambat.
• Watermark yang lebih besar memiliki toleransi tinggi terhadap data yang datang terlambat. Namun, marka air panjang meningkatkan latensi kueri karena kueri harus menyimpan lebih banyak informasi status dan menunggu untuk menulis hasil setelah durasi marka air yang lebih lama.

Tanda air dan mode output untuk agregasi berbingkai

Tabel berikut menunjukkan cara pemrosesan untuk kueri dengan agregasi pada tanda waktu dan watermark:

Marka air dan mode output untuk gabungan stream-stream

Penggabungan beberapa stream hanya mendukung mode append. Kueri akan menulis catatan yang sesuai untuk setiap batch.

Untuk inner join, Databricks menyarankan agar Anda menetapkan ambang watermark pada setiap sumber data streaming agar kueri dapat membuang informasi status untuk record lama. Tanpa marka air, Streaming Terstruktur mencoba menggabungkan setiap kunci dari kedua sisi gabungan pada setiap pemicu, yang dapat memengaruhi performa.

Untuk gabungan luar, marka air adalah wajib. Saat rekaman tidak cocok, kueri menulis nilai null untuk kunci tersebut. Karena gabungan hanya mendukung mode penambahan, rekaman yang tidak cocok tidak ditulis hingga ambang batas keterlambatan berlalu.

Mengendalikan ambang batas data terlambat dengan kebijakan beberapa watermark

Untuk beberapa input Structured Streaming, Anda dapat menetapkan beberapa watermark untuk menentukan ambang toleransi terhadap data yang datang terlambat. Watermark memungkinkan Anda mengendalikan informasi status dan latensi.

Suatu kueri streaming dapat memiliki beberapa aliran input yang disatukan atau digabungkan bersama. Untuk operasi stateful, setiap aliran input mungkin memerlukan ambang batas yang berbeda untuk toleransi terhadap keterlambatan data. Tentukan ambang batas ini menggunakan withWatermark("eventTime", delay) pada setiap aliran input. Berikut ini adalah contoh kueri dengan penggabungan stream-stream.

Python

input_stream1 = ...      # delays up to 1 hour
input_stream2 = ...      # delays up to 2 hours

(input_stream1.withWatermark("eventTime1", "1 hour")
  .join(
    input_stream2.withWatermark("eventTime2", "2 hours"),
    joinCondition)
)

Scala

val inputStream1 = ...      // delays up to 1 hour
val inputStream2 = ...      // delays up to 2 hours

inputStream1.withWatermark("eventTime1", "1 hour")
  .join(
    inputStream2.withWatermark("eventTime2", "2 hours"),
    joinCondition)

Saat menjalankan kueri dengan operasi stateful, Structured Streaming secara terpisah melacak waktu kejadian maksimum untuk setiap aliran input, menghitung watermark berdasarkan penundaan yang sesuai, dan menentukan satu watermark global. Secara bawaan, Structured Streaming menggunakan nilai minimum sebagai watermark global. Jika suatu aliran tertinggal dari aliran lainnya, watermark global minimum mencegah kueri secara keliru menandai data sebagai data terlambat. Misalnya, hal ini dapat terjadi ketika salah satu stream berhenti menerima data karena kegagalan upstream. Watermark global bergerak dengan aman mengikuti laju aliran yang paling lambat dan menunda keluaran kueri bila diperlukan.

Untuk mengurangi latensi, atur spark.sql.streaming.multipleWatermarkPolicy ke max (nilai defaultnya adalah min) untuk menggunakan watermark stream tercepat sebagai watermark global. Namun, konfigurasi ini menghilangkan data dari aliran terlambat. Databricks merekomendasikan agar Anda menerapkan konfigurasi ini dengan hati-hati.

Menerapkan marka air ke operasi yang berbeda

Operasi distinct melacak setiap catatan unik dalam state. Tanpa tanda air, status sistem akan terus bertambah dan dapat menimbulkan masalah memori. Tentukan watermark pada kolom timestamp untuk membatasi status dan menghapus record lama setelah melewati ambang batas.

Contoh berikut menerapkan marka air pada operasi distinct.

Python

streamingDf = spark.readStream. ...  # columns: eventTime, id, value, ...

# Apply watermark before distinct operation
(streamingDf
  .withWatermark("eventTime", "1 hour")
  .distinct()
)

Scala

val streamingDf = spark.readStream. ...  // columns: eventTime, id, value, ...

// Apply watermark before distinct operation
streamingDf
  .withWatermark("eventTime", "1 hour")
  .distinct()

Dalam contoh ini, kueri streaming menghapus rekaman duplikat yang tiba dalam waktu 1 jam dari yang terbaru diamati eventTime. Kueri membuang informasi status untuk deduplikasi setelah ambang batas terlampaui.

Menghilangkan duplikat dalam tanda air

Di Databricks Runtime 13.3 LTS atau yang lebih baru, Anda dapat menggunakan pengenal unik untuk menghapus duplikasi catatan dalam batas watermark.

Streaming Terstruktur menjamin pemrosesan tepat sekali tetapi tidak mendeduplikasi rekaman dari sumber data. Gunakan dropDuplicatesWithinWatermark untuk menghapus duplikat pada bidang apa pun, bahkan ketika bidang berbeda di seluruh rekaman duplikat, seperti waktu peristiwa atau waktu kedatangan.

Dengan dropDuplicatesWithinWatermark, kueri selalu menghapus duplikasi record yang masuk dalam ambang watermark. Kueri juga mungkin menghapus duplikasi catatan yang masuk di luar ambang batas, namun hal ini tidak dijamin. Untuk menjamin kueri membuang semua duplikat, atur ambang watermark agar lebih besar daripada perbedaan stempel waktu maksimum antara event duplikat.

Anda harus menentukan watermark untuk menggunakan metode dropDuplicatesWithinWatermark:

Python

streamingDf = spark.readStream. ...

# deduplicate using guid column with watermark based on eventTime column
(streamingDf
  .withWatermark("eventTime", "10 hours")
  .dropDuplicatesWithinWatermark(["guid"])
)

Scala

val streamingDf = spark.readStream. ...  // columns: guid, eventTime, ...

// deduplicate using guid column with watermark based on eventTime column
streamingDf
  .withWatermark("eventTime", "10 hours")
  .dropDuplicatesWithinWatermark(Seq("guid"))

Gunakan contoh kasus

Contoh berikut menunjukkan kasus penggunaan windowing tingkat lanjut:

Gunakan jendela tumbling untuk menghitung total penjualan per jam

Jendela tumbling berukuran tetap dengan interval yang tidak tumpang tindih. Setiap baris input termasuk dalam tepat satu jendela. Gunakan jendela tumbling untuk menghitung agregasi periode waktu diskrit, seperti total penjualan per jam:

Python

from pyspark.sql.functions import window, sum

hourly_sales = (orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window("timestamp", "1 hour"))
  .agg(sum("amount").alias("total_sales"))
)

Scala

import org.apache.spark.sql.functions.{window, sum}

val hourlySales = orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window($"timestamp", "1 hour"))
  .agg(sum($"amount").alias("total_sales"))

Dalam contoh ini:

• window("timestamp", "1 hour") mengelompokkan pesanan ke interval 1 jam yang tidak tumpang tindih, seperti 05.00 hingga 06.00 dan 06.00 hingga 07.00.
• withWatermark("timestamp", "1 hour") menjaga agregat setiap jendela dalam status hingga tanda waktu akhir jendela lebih lama 1 jam dari tanda waktu pesanan maksimum.

Menggunakan jendela geser untuk menghitung agregat bergulir

Jendela geser berukuran tetap dengan interval yang dapat tumpang tindih. Satu baris dapat menjadi milik beberapa jendela. Gunakan jendela geser untuk menghitung agregat berjalan, seperti penjualan selama periode berjalan 6 jam:

Python

from pyspark.sql.functions import window, sum

rolling_sales = (orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window("timestamp", "6 hours", slideDuration="1 hour"))
  .agg(sum("amount").alias("total_sales"))
)

Scala

import org.apache.spark.sql.functions.{window, sum}

val rollingSales = orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window($"timestamp", "6 hours", "1 hour"))
  .agg(sum($"amount").alias("total_sales"))

Dalam contoh ini:

• window("timestamp", "6 hours", slideDuration="1 hour") mengelompokkan pesanan ke dalam interval 6 jam yang bergeser setiap 1 jam, misalnya: pukul 05.00 hingga 11.00 dan 06.00 hingga 12.00.
• withWatermark("timestamp", "1 hour") menjaga agregat setiap jendela dalam status hingga tanda waktu akhir jendela lebih lama 1 jam dari tanda waktu pesanan maksimum.
• slideDuration harus kurang dari atau sama dengan windowDuration.

Menggunakan jendela sesi untuk memeriksa aktivitas pengguna

Jendela sesi tidak memiliki ukuran tetap. Jendela terbuka saat sebuah baris masuk dan tertutup setelah durasi jeda tanpa baris baru. Gunakan jendela sesi untuk mengagregasi ledakan aktivitas antara periode diam yang panjang, seperti tampilan halaman pengguna dalam periode 30 menit:

Python

from pyspark.sql.functions import session_window, sum

sessionized_page_views = (activity
  .withWatermark("timestamp", "1 hour")
  .groupBy("user_id", session_window("timestamp", gapDuration="30 minutes"))
  .agg(sum("page_views").alias("total_page_views"))
)

Scala

import org.apache.spark.sql.functions.{session_window, sum}

val sessionizedPageViews = activity
  .withWatermark("timestamp", "1 hour")
  .groupBy($"user_id", session_window($"timestamp", "30 minutes"))
  .agg(sum($"page_views").alias("total_page_views"))

Dalam contoh ini:

• session_window("timestamp", gapDuration="30 minutes") membuka jendela saat tampilan halaman pertama tiba. Setiap tampilan halaman berikutnya yang tiba dalam waktu 30 menit memperluas jendela. Jika tidak ada kunjungan halaman dalam 30 menit, periode tersebut berakhir, dan kunjungan halaman berikutnya memulai periode baru.
• withWatermark("timestamp", "1 hour") menjaga agregat setiap sesi dalam status hingga tanda waktu akhir jendela lebih lama 1 jam dari tanda waktu tampilan halaman maksimum.
• Argumen timeColumn untuk window() dan session_window() harus dari TimestampType atau TimestampNTZType.
• Gunakan current_timestamp() untuk menentukan jendela berdasarkan waktu pemrosesan daripada waktu peristiwa.
• Anda dapat mengatur durasi jendela dari mikrodetik hingga hari. Durasi satu bulan atau lebih lama tidak didukung.
• Gunakan mode output complete dengan agregasi berbasis jendela untuk mempertahankan seluruh status jendela tanpa batas waktu. Gunakan mode output append dengan watermark yang sesuai untuk membatasi pertumbuhan state dan mencegah masalah memori pada set data besar. Untuk detail lebih lanjut tentang perilaku mode output, lihat Watermark dan mode output untuk agregasi berjendela.