Асинхронная обработка с помощью transformWithState

Python transformWithState поддерживает асинхронную обработку на asyncioоснове . При одновременном выполнении операций состояния и пользовательской логики по ключам группировки и пакетировании межпроцессного взаимодействия асинхронная обработка обеспечивает более высокую пропускную способность, чем синхронная, при лишь незначительных изменениях кода. Это увеличение пропускной способности не требует сторонних асинхронных библиотек. Продвинутые пользователи могут дополнительно оптимизировать свои приложения с помощью асинхронных шаблонов программирования и библиотек с поддержкой асинхронизации.

Асинхронная обработка поддерживает запросы как в микропакетном, так и в режиме реального времени. Для одного ключа группировки асинхронная обработка сохраняет порядок строк. Разные ключи группировки выполняются параллельно.

Чтобы использовать асинхронную обработку, реализуйте AsyncStatefulProcessor вместо синхронного StatefulProcessor. AsyncStatefulProcessor API повторяет синхронный StatefulProcessor API, поэтому большинству приложений требуются лишь небольшие изменения для использования асинхронного API. См. реализацию AsyncStatefulProcessor.

Для синхронного transformWithState API и основных концепций см. Создайте пользовательское приложение с сохранением состояния transformWithState.

Requirements

Асинхронная обработка предъявляет следующие требования:

  • Доступно в Databricks Runtime 19 и выше, либо на серверных вычислениях с версией окружения 6 и выше.
  • Доступно только для построчного transformWithState API на Python. Асинхронная обработка не поддерживается ни для transformWithStateInPandas, ни для API Scala transformWithState.
  • В конвейере Lakeflow настройте конвейер так, чтобы использовать канал PREVIEW. Это требование применяется как к классическим, так и к бессерверным вычислительным ресурсам конвейера. Канал CURRENT не поддерживает асинхронную обработку. См. channel в разделе «Конфигурации конвейеров» и «каналы выполнения конвейеров Lakeflow».

Реализация AsyncStatefulProcessor

Для преобразования синхронного StatefulProcessor в AsyncStatefulProcessor, внесём следующие изменения:

  • Определите методы API (init, close, handleInputRows, handleExpiredTimer и handleInitialState) с ключевым словом async def.
  • Читайте и обновляйте значения состояния и таймеров с помощью await, либо запускайте их с помощью библиотеки Python asyncio. Это применимо к операциям со состоянием, таким как valueState.get() и к операциям с таймером, например registerTimer. Создание объектов состояний, таких как handle.getValueState, остаётся синхронным.

К асинхронной обработке применимы следующие соображения:

  • Если ваше приложение хранит данные в переменных-членах или во внешних системах, Databricks рекомендует переписать логику для безопасного параллельного выполнения. Поскольку handleInputRows и handleExpiredTimer могут одновременно выполняться для разных ключей группировки, чередующиеся запуски не должны повреждать общие данные. Большинство приложений уже соответствуют этому требованию.
  • Databricks рекомендует не перехватывать и не подавлять ошибки при операциях состояния. Apache Spark обрабатывает эти ошибки за вас. Если операция состояния не удаётся, Apache Spark завершает задачу с ошибкой и повторяет её.
    • В AsyncStatefulProcessorсостоянии ошибки операции управляются за вас и никогда не передаются вашему коду.
    • В синхронном StatefulProcessorрежиме в вашем коде возникают ошибки операций с состоянием, но их подавление может нарушить корректность данных.

Пример: подсчёт строк для каждого ключа группировки

Следующий пример определяет AsyncCountProcessor, который считает количество строк для каждого ключа группировки. Переменная value_schema определяет схему ValueState, которая хранит накопительный счётчик. По сравнению с синхронным StatefulProcessor, изменения — это ключевое слово async def в каждом методе и await в операциях чтения и обновления состояния. Вызов getValueState в init остаётся синхронным. Определим процессор как в следующем коде:

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

Выполните запрос с асинхронным процессором

Чтобы выполнить запрос с асинхронным процессором, передайте AsyncStatefulProcessor методу transformWithState. Запрос использует тот же синтаксис, что и синхронный маршрут. Асинхронные и синхронные API имеют одинаковый формат состояния, поэтому вы можете переключать существующий запрос между AsyncStatefulProcessor и синхронным StatefulProcessor, повторно используя одну и ту же контрольную точку.

Пример: подсчёт событий в наборе данных events sample

Следующий пример запускает AsyncCountProcessor на events выборочном наборе данных. Каждая запись имеет поле time (эпохальные секунды) и action поле со значением Open или Close. Запрос группирует по action и считает события для каждого типа действия. Для получения дополнительных выборочных наборов данных см. Примеры наборов данных.

Переменная input_schema определяет схему исходных записей, а переменная output_schema — схему строк, которые выдаёт процессор. Чтобы прочитать пример набора данных как поток, определите обе схемы, затем запустите запрос как в следующем коде:

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()

После завершения запроса просмотрите текущий счёт для каждого типа действия, как в следующем примере кода:

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

Асинхронные состояния и операции с таймером

В AsyncStatefulProcessor операции с переменной состояния и таймером, которые считывают или записывают значения, асинхронны. Большинство этих операций возвращают один результат, который вы получаете с помощью await. Операции, возвращающие коллекцию, вместо этого возвращают асинхронный итератор, который вы потребляете с помощью async for. Введение в async/await и асинхронные итераторы в Python см. в документации Python asyncio.

В следующей таблице перечислены операции, которые возвращают один результат, который можно получить с помощью await:

Class Операции, использующие await
AsyncValueState exists, get, update, clear
AsyncMapState exists, getValue, containsKey, updateValue, removeKey, clear
AsyncListState exists, put, appendValue, appendList, clear
AsyncStatefulProcessorHandle registerTimer, deleteTimer

В следующей таблице перечислены операции, возвращающие асинхронный итератор, который можно получить с помощью async for:

Class Операции, использующие async for
AsyncMapState iterator, keys, values
AsyncListState get
AsyncStatefulProcessorHandle listTimers

Пример: async for

Например, чтобы прочитать значения в AsyncListState, выполните итерацию по нему с помощью async for, как в следующем коде:

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

Методы, создающие объекты состояния и удаляющие переменные состояния, остаются синхронными: getValueState, getMapState, getListState, и deleteIfExists.

Описание каждого типа состояния см. в разделе «Пользовательские типы состояний».

Оптимизируйте с помощью асинхронных шаблонов программирования

Асинхронная обработка полезна, когда ваша логика ждёт внешних операций, таких как сетевые запросы. Вместо того чтобы ждать каждого запроса последовательно, используйте asyncio, чтобы выполнять запросы одновременно и сократить время простоя.

Пример: выполните параллельные запросы с asyncio.gather

В следующем примере asyncio.gather используется для одновременного запуска всех HTTP-запросов для каждой строки и ожидания их завершения, после чего максимальный балл сохраняется в состоянии. Определим процессор как в следующем коде:

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()
  • Последнее обновление: