Pemrosesan asinkron dengan transformWithState

Python transformWithState mendukung pemrosesan asinkron yang dibangun di atas asyncio. Dengan menjalankan operasi state dan logika pengguna secara bersamaan di seluruh grouping key serta menggabungkan komunikasi antar-proses, pemrosesan asinkron memiliki throughput lebih tinggi dibandingkan pemrosesan sinkron dengan hanya perubahan kode minor. Peningkatan throughput ini tidak memerlukan pustaka asinkron pihak ketiga. Pengguna tingkat lanjut dapat mengoptimalkan aplikasi mereka lebih lanjut dengan pola pemrograman async dan pustaka yang mendukung async.

Pemrosesan asinkron mendukung kueri mode baik mikrobatch maupun real-time. Untuk satu kunci pengelompokan, pemrosesan asinkron mempertahankan urutan baris. Proses dengan kunci pengelompokan yang berbeda berjalan secara bersamaan.

Untuk menggunakan pemrosesan asinkron, implementasikan AsyncStatefulProcessor alih-alih StatefulProcessor sinkron. API AsyncStatefulProcessor ini mencerminkan API sinkron StatefulProcessor, sehingga sebagian besar aplikasi hanya memerlukan perubahan kecil untuk menggunakan API asinkron. Lihat Implement AsyncStatefulProcessor.

Untuk API sinkron transformWithState dan konsep inti, lihat Membangun aplikasi stateful kustom dengan transformWithState.

Requirements

Pemrosesan asinkron memiliki persyaratan sebagai berikut:

  • Tersedia di Databricks Runtime 19 dan ke atas, atau pada komputasi serverless dengan versi lingkungan 6 ke atas.
  • Tersedia hanya untuk API Python berbasis baris transformWithState. Pemrosesan asinkron tidak didukung untuk transformWithStateInPandas atau untuk API Scala transformWithState .
  • Dalam pipeline Lakeflow, konfigurasikan pipeline untuk menggunakan channel PREVIEW. Persyaratan ini berlaku untuk komputasi pipeline klasik dan serverless. Kanal CURRENT tidak mendukung pemrosesan asinkron. Lihat channel di Konfigurasi pipeline dan saluran runtime Lakeflow pipelines.

Implementasikan AsyncStatefulProcessor

Untuk mengonversi objek sinkron StatefulProcessor menjadi AsyncStatefulProcessor, lakukan perubahan berikut:

  • Definisikan metode API (init, close, handleInputRows, handleExpiredTimer, dan handleInitialState) dengan async def kata kunci.
  • Baca dan perbarui nilai status dan timer dengan await, atau menjalankannya menggunakan pustaka Python asyncio. Ini berlaku untuk operasi status seperti valueState.get() dan untuk operasi timer seperti registerTimer. Membuat objek state, seperti handle.getValueState, tetap sinkron.

Pertimbangan berikut berlaku untuk pemrosesan asinkron:

  • Jika aplikasi Anda menyimpan data dalam variabel anggota atau di sistem eksternal, Databricks merekomendasikan agar Anda menulis ulang logika agar aman untuk eksekusi bersamaan. Karena handleInputRows dan handleExpiredTimer dapat berjalan bersamaan di seluruh kunci pengelompokan, run interleaved tidak boleh merusak data yang dibagikan. Sebagian besar aplikasi sudah memenuhi persyaratan ini.
  • Databricks menyarankan agar Anda tidak menangkap atau mengabaikan error dari operasi state. Apache Spark menangani kesalahan ini untuk Anda. Jika operasi state gagal, Apache Spark membuat tugas tersebut gagal dan mencobanya kembali.
    • Dalam AsyncStatefulProcessor, kesalahan operasi state dikelola untuk Anda dan tidak pernah diteruskan ke kode Anda.
    • Dalam operasi sinkron, StatefulProcessorkesalahan operasi state muncul dalam kode Anda, tetapi menekan kesalahan tersebut dapat mengorbankan kebenaran data.

Contoh: hitung baris untuk setiap kunci pengelompokan

Contoh berikut mendefinisikan AsyncCountProcessor yang menghitung jumlah baris untuk setiap kunci pengelompokan. Variabel value_schema mendefinisikan skema objek ValueState yang menyimpan jumlah yang berjalan. Dibandingkan dengan StatefulProcessor sinkron, perubahan tersebut adalah kata kunci async def pada setiap metode dan await pada operasi pembacaan dan pembaruan status. Panggilan ke getValueState di init tetap sinkron. Definisikan prosesor seperti dalam kode berikut:

from pyspark.sql import Row
from pyspark.sql.streaming import AsyncStatefulProcessor, AsyncStatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, LongType

value_schema = StructType([StructField("count", LongType(), True)])

class AsyncCountProcessor(AsyncStatefulProcessor):
  async def init(self, handle: AsyncStatefulProcessorHandle) -> None:
    self.count = handle.getValueState("count", value_schema)

  async def handleInputRows(self, key, rows, timerValues):
    total = (await self.count.get() or (0,))[0]
    for _ in rows:
      total += 1
    await self.count.update((total,))
    yield Row(action=key[0], count=total)

  async def close(self) -> None:
    pass

Jalankan kueri dengan prosesor asinkron

Untuk menjalankan query dengan prosesor asinkron, berikan AsyncStatefulProcessor ke transformWithState. Kueri menggunakan sintaks yang sama dengan jalur sinkron. API asinkron dan sinkron berbagi format state yang sama, sehingga Anda dapat mengganti kueri yang sudah ada antara AsyncStatefulProcessor dan StatefulProcessor sinkron sambil menggunakan kembali checkpoint yang sama.

Contoh: hitung peristiwa dalam events dataset contoh

Contoh berikut menjalankan AsyncCountProcessor pada dataset sampel events. Setiap catatan memiliki bidang time (detik epoch) dan bidang action dengan nilai Open atau Close. Query mengelompokkan berdasarkan action dan menghitung event untuk setiap jenis aksi. Untuk contoh dataset lainnya, lihat Dataset contoh.

input_schema variabel mendefinisikan skema catatan sumber, dan output_schema variabel mendefinisikan skema baris yang dihasilkan prosesor. Untuk membaca dataset contoh sebagai aliran, definisikan kedua skema, lalu mulai kueri seperti dalam kode berikut:

from pyspark.sql.types import StructType, StructField, StringType, LongType

input_schema = StructType([
  StructField("time", LongType(), True),
  StructField("action", StringType(), True),
])

output_schema = StructType([
  StructField("action", StringType(), True),
  StructField("count", LongType(), True),
])

events = (
  spark.readStream.schema(input_schema)
    .option("maxFilesPerTrigger", 10)
    .json("/databricks-datasets/structured-streaming/events")
)

q = (
  events.groupBy("action")
    .transformWithState(
      statefulProcessor=AsyncCountProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
    )
    .writeStream.format("memory")
    .queryName("async_counts")
    .trigger(availableNow=True)
    .start()
)

q.awaitTermination()

Setelah kueri selesai, lihat jumlah berjalan untuk setiap jenis aksi seperti dalam kode berikut:

display(spark.sql("SELECT action, MAX(count) AS count FROM async_counts GROUP BY action ORDER BY action"))

Operasi status asinkron dan timer

Dalam AsyncStatefulProcessor, operasi variabel status dan timer yang membaca atau menulis nilai bersifat asinkron. Sebagian besar operasi ini mengembalikan satu hasil yang Anda ambil dengan await. Operasi yang mengembalikan koleksi justru mengembalikan iterator asinkron yang Anda konsumsi dengan async for. Untuk pengantar tentang async/await dan iterator asinkron di Python, lihat dokumentasi Python asyncio.

Tabel berikut mencantumkan operasi yang mengembalikan satu hasil yang dapat Anda ambil dengan await:

Class Operasi yang menggunakan await
AsyncValueState exists,get,update,clear
AsyncMapState exists, getValue, containsKey, updateValue, removeKey, clear
AsyncListState exists, put, appendValue, appendList, clear
AsyncStatefulProcessorHandle registerTimer, deleteTimer

Tabel berikut mencantumkan operasi yang mengembalikan iterator asinkron yang dapat Anda ambil dengan async for:

Class Operasi yang menggunakan async for
AsyncMapState iterator, keys, values
AsyncListState get
AsyncStatefulProcessorHandle listTimers

Contoh: async for

Misalnya, untuk membaca nilai dalam AsyncListState, lakukan iterasi dengan async for seperti dalam kode berikut:

total = 0
async for value in self.items.get():
  total += value[0]

Metode yang membuat objek status dan menghapus variabel status tetap sinkron: getValueState, getMapState, getListState, dan deleteIfExists.

Untuk deskripsi setiap jenis state, lihat Jenis state kustom.

Optimasi dengan pola pemrograman asinkron

Pemrosesan asinkron berguna ketika logika Anda bergantung pada operasi eksternal, seperti permintaan jaringan. Alih-alih menunggu setiap permintaan secara berurutan, gunakan asyncio untuk menjalankan permintaan secara bersamaan dan mengurangi waktu idle.

Contoh: jalankan request bersamaan dengan asyncio.gather

Contoh berikut menggunakan asyncio.gather untuk menjalankan semua permintaan HTTP per baris secara bersamaan dan menunggu hingga selesai, lalu menyimpan skor maksimum dalam status. Definisikan prosesor seperti dalam kode berikut:

import asyncio
import aiohttp
from pyspark.sql import Row
from pyspark.sql.streaming import AsyncStatefulProcessor

class HttpScoreRowGatherProcessor(AsyncStatefulProcessor):
  async def init(self, handle):
    self._score_state = handle.getValueState("last_score", "score double")
    self._session = aiohttp.ClientSession()

  async def _fetch_score(self, row) -> float:
    async with self._session.get(
      f"https://api.example.com/score/{row.event_id}"
    ) as resp:
      return (await resp.json())["score"]

  async def handleInputRows(self, key, rows, timerValues):
    user_id = key[0]
    scores = await asyncio.gather(*[self._fetch_score(row) for row in rows])

    max_score = max(scores)
    await self._score_state.update((max_score,))
    yield Row(user_id=user_id, score=max_score)

  async def close(self):
    await self._session.close()
  • Terakhir diperbarui pada