Window bergulir di Streaming Terstruktur

Jendela bergulir menghitung agregasi berbasis waktu di sepanjang frontier yang terus berkembang. Berbeda dengan jendela tumbling atau sliding yang menetapkan setiap event ke bucket waktu tetap, rolling window mengeluarkan agregat terbaru saat frontier bergerak maju dan event lama keluar dari jangkauan. Gunakan rolling window untuk penyajian fitur secara real-time dan pemantauan berlatensi rendah, ketika Anda memerlukan nilai terkini dari suatu agregat dalam jendela lookback terbaru.

Jendela rolling mendukung hal-hal berikut:

  • Window berbatas yang beragregasi selama durasi lookback tetap, seperti 10 menit terakhir.
  • Jendela seumur hidup yang mengumpulkan semua data sejak aliran dimulai.
  • Beberapa jendela dalam satu operator, sehingga Anda dapat menghitung durasi lookback yang berbeda dalam satu panggilan.

Requirements

  • Databricks Runtime 19 ke atas.
  • Mode mikro-batch: komputasi klasik (mode akses standar atau khusus) atau komputasi tanpa server.
  • Mode waktu nyata: komputasi klasik (mode akses standar atau khusus). Dalam pipeline Lakeflow, mode real-time juga berjalan di komputasi serverless melalui konfigurasi pipeline. Lihat Menggunakan mode real time di alur Lakeflow.

Untuk informasi selengkapnya, lihat mode Eksekusi.

Cara kerja rolling window

Query rolling window mengelompokkan peristiwa berdasarkan satu atau lebih kolom partisi, mengurutkannya berdasarkan kolom timestamp, dan menghitung agregat relatif terhadap sebuah frontier. Setiap baris keluaran menyertakan nilai frontier sebagai kolom timestamp.

Garis Depan

Batas menandai "waktu kini" saat ini bagi operator. Ini berjalan berdasarkan waktu pemrosesan (wall clock), dan secara opsional dapat tertinggal sebesar delay yang dapat dikonfigurasi. Penundaan berguna untuk menyerap sedikit keterlambatan ingesti sehingga event yang sedang diproses dihitung sebelum frontier pemrosesan melewatinya.

Definisikan batas dengan FrontierSpec:

# Frontier with a 5-second delay, emitted as a column named "window_time"
RollingWindow.FrontierSpec(delay="5 seconds", alias="window_time")

Jendela berbatas

Window berbatas mengumpulkan data dalam durasi tertentu, melihat ke belakang dari frontier. Definisikan lookback dengan .over():

# Sum of "value" over the last hour
F.sum("value").over(RollingWindow.preceding(RollingWindow.Range("1 hour")))

Jendela seumur sesi

Jendela sepanjang masa aktif mengumpulkan data dari awal aliran tanpa kedaluwarsa. Untuk membuat jendela seumur hidup, hilangkan :.over()

# Running count of non-null values since the stream started
F.count("value")

Mode eksekusi

Jendela rolling berjalan dalam mode micro-batch dan mode real-time. Frontier berperilaku berbeda di masing-masing:

  • Mode mikro-batch: Frontier ditetapkan untuk setiap mikro-batch. Setiap kunci partisi menghasilkan paling banyak satu baris output per batch.
  • Mode real-time: Batas depan bergerak maju terus-menerus selama satu batch. Kunci partisi dapat mengeluarkan beberapa baris output dalam satu batch saat frontier bergerak.

Kejadian terlambat dan watermark

Jendela bergulir mengatur batas jendela dari frontier waktu pemrosesan, tetapi Anda dapat menerapkan watermark waktu peristiwa untuk membuang event yang terlambat datang. Tambahkan .withWatermark() sebelum rollingWindow(), menggunakan kolom yang sama yang Anda teruskan ke orderBy:

df.withWatermark("event_time", "5 seconds").rollingWindow(
    partitionBy="user_id",
    orderBy="event_time",
    frontierSpec=RollingWindow.FrontierSpec(delay="0 seconds"),
    measures=[F.sum("value").over(RollingWindow.preceding(RollingWindow.Range("10 minutes")))],
)

Event yang waktu eventnya berada di atau di bawah watermark akan dihapus, bahkan ketika berada dalam rentang waktu jendela. Tanpa watermark, setiap peristiwa yang masuk dalam jendela akan diagregasi, termasuk kedatangan terlambat. Watermark berlaku baik dalam mode micro-batch maupun real-time, dan posisi watermark disimpan dalam checkpoint, sehingga dipulihkan saat query dimulai ulang.

Syntax

Panggil rollingWindow() pada DataFrame streaming:

Python

from pyspark.sql.streaming.rolling_window import RollingWindow
from pyspark.sql import functions as F

df.rollingWindow(
    partitionBy="user_id",
    orderBy="event_time",
    frontierSpec=RollingWindow.FrontierSpec(
        delay="0 seconds",
        alias="frontier",
    ),
    measures=[
        F.sum("value").over(RollingWindow.preceding(RollingWindow.Range("10 minutes"))),
        F.count("value").alias("non_null_value_count"),  # No .over() = lifetime window
    ],
)

Scala

import org.apache.spark.sql.streaming.RollingWindow
import org.apache.spark.sql.streaming.RollingWindow.{FrontierSpec, Range, preceding}
import org.apache.spark.sql.functions._

df.rollingWindow(
  partitionBy = Seq("user_id"),
  orderBy = "event_time",
  frontierSpec = FrontierSpec(delay = "0 seconds", alias = "frontier"),
  measures = Seq(
    sum("value").over(preceding(Range("10 minutes"))),
    count("value")
  )
)

Operator rollingWindow() mengambil parameter berikut:

Parameter Description
partitionBy Required. Satu atau lebih kolom untuk mengelompokkan peristiwa.
orderBy Required. Kolom timestamp yang mengurutkan event. Harus berupa kolom TimestampType.
frontierSpec Definisi frontier. Atur delay untuk mengontrol seberapa jauh frontier tertinggal dari waktu pemrosesan, dan alias untuk menamai kolom output frontier.
measures Daftar ekspresi agregat. Tambahkan .over() untuk window dengan rentang terbatas, atau hilangkan untuk window berlaku seumur hidup.

Fungsi agregat yang didukung

Rolling window mendukung fungsi agregat berikut:

  • sum
  • count
  • avg
  • min
  • max
  • stddev, termasuk varian sampel (stddev_samp) dan populasi (stddev_pop)
  • variance, termasuk varian sampel (var_samp) dan populasi (var_pop)
  • approx_count_distinct
  • approx_percentile
  • UDAF Scala Aggregator yang didaftarkan dengan functions.udaf. Didukung hanya pada classic compute. Sebagai contoh, lihat Contoh 3: Scala UDAF.

Examples

Contoh 1: Pendapatan rolling per pengguna selama satu jam terakhir

Contoh berikut menghitung jumlah pendapatan bergulir per pengguna selama satu jam terakhir dan menulis pembaruan ke Kafka:

from pyspark.sql.streaming.rolling_window import RollingWindow
from pyspark.sql import functions as F

result = events_df.rollingWindow(
    partitionBy="user_id",
    orderBy="event_time",
    frontierSpec=RollingWindow.FrontierSpec(delay="0 seconds"),
    measures=[
        F.sum("revenue")
        .over(RollingWindow.preceding(RollingWindow.Range("1 hour")))
        .alias("rolling_revenue"),
    ],
)

(
    result.select(
        F.col("user_id").cast("string").alias("key"),
        F.to_json(F.struct("__frontier", "rolling_revenue")).alias("value"),
    )
    .writeStream.format("kafka")
    .option("kafka.bootstrap.servers", "<host1:port1,host2:port2>")
    .option("topic", "<topic-name>")
    .outputMode("update")
    .start()
)

Contoh 2: Jendela terbatas campuran dan jangka waktu permanen

Contoh berikut menghitung dua agregat terbatas dan satu agregat sepanjang masa hidup dalam satu operator:

result = events_df.rollingWindow(
    partitionBy=["region", "product"],
    orderBy="event_time",
    frontierSpec=RollingWindow.FrontierSpec(delay="10 seconds", alias="window_ts"),
    measures=[
        F.sum("revenue").over(RollingWindow.preceding(RollingWindow.Range("1 hour"))),
        F.avg("price").over(RollingWindow.preceding(RollingWindow.Range("10 minutes"))),
        F.count("value"),  # Lifetime: non-null values since the stream started
    ],
)

Skema keluaran adalah (region, product, window_ts, sum(revenue), avg(price), count(value)).

Contoh 3: Scala UDAF

Note

UDAF di Scala Aggregator hanya didukung pada komputasi klasik.

Contoh berikut mendefinisikan rata-rata kustom sebagai Scala Aggregator, mendaftarkannya dengan udaf, dan menggunakannya sebagai ukuran selama jendela terbatas 10 menit:

import org.apache.spark.sql.{Encoder, Encoders}
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.RollingWindow.{FrontierSpec, Range, preceding}

case class AvgBuffer(sum: Long, count: Long)

object CustomAverage extends Aggregator[Long, AvgBuffer, java.lang.Double] {
  override def zero: AvgBuffer = AvgBuffer(0L, 0L)
  override def reduce(buffer: AvgBuffer, value: Long): AvgBuffer =
    AvgBuffer(buffer.sum + value, buffer.count + 1)
  override def merge(left: AvgBuffer, right: AvgBuffer): AvgBuffer =
    AvgBuffer(left.sum + right.sum, left.count + right.count)
  override def finish(buffer: AvgBuffer): java.lang.Double =
    if (buffer.count == 0) null else buffer.sum.toDouble / buffer.count
  override def bufferEncoder: Encoder[AvgBuffer] = Encoders.product
  override def outputEncoder: Encoder[java.lang.Double] = Encoders.DOUBLE
}

val customAvg = udaf(CustomAverage)

val result = events.rollingWindow(
  partitionBy = Seq("user_id"),
  orderBy = "event_time",
  frontierSpec = FrontierSpec(delay = "0 seconds", alias = "frontier"),
  measures = Seq(
    customAvg(col("value").cast("long"))
      .over(preceding(Range("10 minutes")))
      .as("custom_avg_10m")
  )
)

Limitations

  • Gunakan update mode output. Mode output append dan complete tidak didukung.
  • Setiap measure dapat memiliki paling banyak satu .over() klausa. Anda tidak bisa merangkai spesifikasi jendela.
  • Anda tidak bisa menggabungkan rolling window dengan fungsi jendela Spark biasa. Ekspresi satu kolom tidak dapat menggunakan baik .over(WindowSpec) maupun .over(RollingWindow...).
  • Frontier hanya bertambah berdasarkan waktu pemrosesan. Batas waktu peristiwa tidak didukung.
  • UDAF Scala Aggregator hanya didukung pada komputasi klasik.
  • Databricks merekomendasikan menggunakan durasi rolling window kurang dari atau sama dengan 7 hari. Durasi lebih dari 7 hari mungkin memerlukan penyimpanan eksekutor yang besar dan menyebabkan kinerja lambat atau kegagalan klaster.
  • Terakhir diperbarui pada