Использовать потоки в конвейерах Lakeflow

Потоки в конвейере Lakeflow перемещают данные в потоковую таблицу или материализованное представление. В следующих примерах показано, как определить потоки по умолчанию, определить поток отдельно от целевого ресурса, записывать в потоковую таблицу данные из нескольких топиков Kafka, выполнить однократное обратное заполнение и заменить запросы UNION обработкой потока в режиме добавления.

Обзор потоков см. в разделе Поэтапная загрузка и обработка данных с помощью потоков конвейера Lakeflow.

Пример. Создание потока по умолчанию

При создании конвейера обычно определяется таблица или представление вместе с запросом, поддерживающим его. Например, этот запрос создает потоковую таблицу с именем customers_silver, считывая данные из customers_bronze. Потоковая таблица и её поток по умолчанию создаются одновременно за один шаг.

SQL

CREATE OR REFRESH STREAMING TABLE customers_silver
AS SELECT * FROM STREAM(customers_bronze)

Питон

from pyspark import pipelines as dp

@dp.table()
def customers_silver():
  return spark.readStream.table("customers_bronze")

Режим по умолчанию для потоковой таблицы — это режим добавления, который добавляет новые строки при каждом обновлении и имеет то же имя, что и целевая таблица. Это наиболее распространенный способ использования конвейеров — создание потока и его целевого объекта на одном шаге, и его можно использовать для приема или преобразования данных. Дополнительные сведения о концепциях потоков см. в разделе Поэтапная загрузка и обработка данных с помощью потоков конвейера Lakeflow.

Пример. Определение потока отдельно от целевого объекта

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

Питон

from pyspark import pipelines as dp

# create streaming table
dp.create_streaming_table("customers_silver")

# add a flow
@dp.append_flow(
  target = "customers_silver")
def customer_silver():
  return spark.readStream.table("customers_bronze")

SQL

-- create a streaming table
CREATE OR REFRESH STREAMING TABLE customers_silver;

-- add a flow
CREATE FLOW customers_silver
AS INSERT INTO customers_silver BY NAME
SELECT * FROM STREAM(customers_bronze);

Определение потока отдельно от целевого объекта позволяет создавать несколько потоков, добавляющих данные в один целевой объект. Используйте декоратор @dp.append_flow в интерфейсе Python или оператор CREATE FLOW...INSERT INTO в интерфейсе SQL, чтобы добавлять потоки для таких задач, как следующие:

Добавьте источники потоковой передачи, добавляющие данные в существующую потоковую таблицу без полного обновления. Например, у вас может быть таблица, объединяющая региональные данные из каждого региона, в который вы работаете. При развертывании новых регионов можно добавить новые данные региона в таблицу без полного обновления. См. пример: запись в потоковую таблицу из нескольких топиков Kafka.
Добавьте отсутствующие исторические данные (обратное заполнение) в стриминговую таблицу. Вы можете использовать синтаксис INSERT INTO ONCE для создания исторического дозаполнения, которое выполняется однократно. См. Пример: однократное дозаполнение данных и Дозаполнение исторических данных с помощью конвейеров.
Объедините данные из нескольких источников и запишите в одну потоковую таблицу вместо использования оператора UNION в запросе. Использование обработки потока добавления вместо UNION позволяет постепенно обновлять целевую таблицу без выполнения полного обновления. См. пример. Используйте обработку потока добавления вместо UNION.

Для запросов Python используйте функцию create_streaming_table() для создания целевой таблицы.

Это важно

Если необходимо определить ограничения качества данных с ожиданиями, определите ожидания в целевой таблице в рамках create_streaming_table() функции или существующего определения таблицы. Вы не можете определить ожидания в определении @append_flow .
Потоки определяются именем потока, и это имя используется для идентификации контрольных точек потоковой передачи. Использование имени потока для идентификации контрольной точки означает следующее:
- Если существующий поток в конвейере переименован, контрольная точка не переносится, и переименованный поток фактически является совершенно новым потоком.
- Невозможно повторно использовать имя потока в конвейере, так как существующая контрольная точка не будет соответствовать новому определению потока.

Пример. Запись в потоковую таблицу из нескольких разделов Kafka

В следующих примерах создается таблица потоковой передачи с именем kafka_target, и выполняется запись в нее из двух топиков Kafka.

Питон

from pyspark import pipelines as dp

dp.create_streaming_table("kafka_target")

# Kafka stream from multiple topics
@dp.append_flow(target = "kafka_target")
def topic1():
  return (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,...")
      .option("subscribe", "topic1")
      .load()
  )

@dp.append_flow(target = "kafka_target")
def topic2():
  return (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,...")
      .option("subscribe", "topic2")
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE kafka_target;

CREATE FLOW
  topic1
AS INSERT INTO
  kafka_target BY NAME
SELECT * FROM
  read_kafka(bootstrapServers => 'host1:port1,...', subscribe => 'topic1');

CREATE FLOW
  topic2
AS INSERT INTO
  kafka_target BY NAME
SELECT * FROM
  read_kafka(bootstrapServers => 'host1:port1,...', subscribe => 'topic2');

Дополнительные сведения о табличной функции, используемой в SQL-запросах, см. в справочнике по языку SQL в разделе read_kafka().

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

Замечание

Паттерн имеет те же требования, что и цикл for, используемый для создания таблиц. Необходимо явно передать значение Python в функцию, определяющую поток. См. статью "Создание таблиц в циклеfor".

from pyspark import pipelines as dp

dp.create_streaming_table("kafka_target")

topic_list = ["topic1", "topic2", "topic3"]

for topic_name in topic_list:

  @dp.append_flow(target = "kafka_target", name=f"{topic_name}_flow")
  def topic_flow(topic=topic_name):
    return (
      spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "host1:port1,...")
        .option("subscribe", topic)
        .load()
    )

Пример: Запуск одноразового заполнения данных

Если вы хотите запустить запрос, чтобы добавить данные в существующую потоковую таблицу, используйте append_flow.

После добавления набора существующих данных у вас есть несколько параметров:

Если вы хотите, чтобы запрос добавлял новые данные, поступающие в директорию для догрузки данных, оставьте запрос в его текущем виде.
Если вы хотите, чтобы это было однократное резервное заполнение, и никогда не выполняйте его снова, удалите запрос после запуска конвейера один раз.
Если вы хотите, чтобы запрос выполнился один раз и повторно выполнялся только в тех случаях, когда данные полностью обновляются, установите параметр once в положение True в процессе добавления. В SQL используйте INSERT INTO ONCE.

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

Питон

from pyspark import pipelines as dp

@dp.table()
def csv_target():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format","csv")
    .load("path/to/sourceDir")

@dp.append_flow(
  target = "csv_target",
  once = True)
def backfill():
  return spark.read
    .format("cloudFiles")
    .option("cloudFiles.format","csv")
    .load("path/to/backfill/data/dir")

SQL

CREATE OR REFRESH STREAMING TABLE csv_target
AS SELECT * FROM
  read_files(
    "path/to/sourceDir",
    "csv"
  );

CREATE FLOW
  backfill
AS INSERT INTO ONCE
  csv_target BY NAME
SELECT * FROM
  read_files(
    "path/to/backfill/data/dir",
    "csv"
  );

Более подробный пример см. в статье "Обратное заполнение исторических данных с помощью пайплайнов".

Пример: Используйте обработку потока добавления вместо `UNION`

Вместо использования запроса с UNION предложением, вы можете использовать запросы на добавление потоков для объединения нескольких источников и записи в одну потоковую таблицу. Использование запросов на добавление вместо UNION позволяет добавлять данные в потоковую таблицу из разных источников без полного обновления.

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

@dp.create_table(name="raw_orders")
def unioned_raw_orders():
  raw_orders_us = (
    spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/path/to/orders/us")
  )

  raw_orders_eu = (
    spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/path/to/orders/eu")
  )

  return raw_orders_us.union(raw_orders_eu)

Следующие примеры показывают замену запроса UNION на запросы потока добавления.

Питон

dp.create_streaming_table("raw_orders")

@dp.append_flow(target="raw_orders")
def raw_orders_us():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .load("/path/to/orders/us")

@dp.append_flow(target="raw_orders")
def raw_orders_eu():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .load("/path/to/orders/eu")

# Additional flows can be added without the full refresh that a UNION query would require:
@dp.append_flow(target="raw_orders")
def raw_orders_apac():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .load("/path/to/orders/apac")

SQL

CREATE OR REFRESH STREAMING TABLE raw_orders;

CREATE FLOW
  raw_orders_us
AS INSERT INTO
  raw_orders BY NAME
SELECT * FROM
  STREAM read_files(
    "/path/to/orders/us",
    format => "csv"
  );

CREATE FLOW
  raw_orders_eu
AS INSERT INTO
  raw_orders BY NAME
SELECT * FROM
  STREAM read_files(
    "/path/to/orders/eu",
    format => "csv"
  );

-- Additional flows can be added without the full refresh that a UNION query would require:
CREATE FLOW
  raw_orders_apac
AS INSERT INTO
  raw_orders BY NAME
SELECT * FROM
  STREAM read_files(
    "/path/to/orders/apac",
    format => "csv"
  );

Пример. Использование `transformWithState` для мониторинга пульса датчика

В следующем примере показан обработчик с сохранением состояния, который считывает данные из Kafka и проверяет, что датчики периодически посылают сигналы heartbeat. Если сигнал не поступает в течение 5 минут, процессор вносит запись в целевую таблицу Delta для анализа.

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

Замечание

RocksDB — это поставщик состояний по умолчанию, начиная с Databricks Runtime 17.2. Если запрос завершается ошибкой из-за исключения провайдера, не поддерживаемого, добавьте следующие конфигурации конвейера, выполните полное обновление или сброс этих чекпоинтов, а затем повторно запустите конвейер.

"configuration": {
    "spark.sql.streaming.stateStore.providerClass": "com.databricks.sql.streaming.state.RocksDBStateStoreProvider",
    "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled": "true"
}

from typing import Iterator

import pandas as pd

from pyspark import pipelines as dp
from pyspark.sql.functions import col, from_json
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, LongType, StringType, TimestampType

KAFKA_TOPIC = "<your-kafka-topic>"

output_schema = StructType([
    StructField("sensor_id", LongType(), False),
    StructField("sensor_type", StringType(), False),
    StructField("last_heartbeat_time", TimestampType(), False)])

class SensorHeartbeatProcessor(StatefulProcessor):
    def init(self, handle: StatefulProcessorHandle) -> None:
        # Define state schema to store sensor information (sensor_id is the grouping key)
        state_schema = StructType([
            StructField("sensor_type", StringType(), False),
            StructField("last_heartbeat_time", TimestampType(), False)])
        self.sensor_state = handle.getValueState("sensorState", state_schema)
        # State variable to track the previously registered timer
        timer_schema = StructType([StructField("timer_ts", LongType(), False)])
        self.timer_state = handle.getValueState("timerState", timer_schema)
        self.handle = handle

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        # Process one row from input and update state
        pdf = next(rows)
        row = pdf.iloc[0]
        # Store or update the sensor information in state using current timestamp
        current_time = pd.Timestamp(timerValues.getCurrentProcessingTimeInMs(), unit='ms')
        self.sensor_state.update((
            row["sensor_type"],
            current_time
        ))

        # Delete old timer if already registered
        if self.timer_state.exists():
            old_timer = self.timer_state.get()[0]
            self.handle.deleteTimer(old_timer)

        # Register a timer for 5 minutes from current processing time
        expiry_time = timerValues.getCurrentProcessingTimeInMs() + (5 * 60 * 1000)
        self.handle.registerTimer(expiry_time)
        # Store the new timer timestamp in state
        self.timer_state.update((expiry_time,))

        # No output on input processing, output only on timer expiry
        return iter([])

    def handleExpiredTimer(self, key, timerValues, expiredTimerInfo) -> Iterator[pd.DataFrame]:
        # Emit output row based on state store
        if self.sensor_state.exists():
            state = self.sensor_state.get()
            output = pd.DataFrame({
                "sensor_id": [key[0]],  # Use grouping key as sensor_id
                "sensor_type": [state[0]],
                "last_heartbeat_time": [state[1]]
            })
            # Remove the entry for the sensor from the state store
            self.sensor_state.clear()
            # Remove the timer state entry
            self.timer_state.clear()
            yield output

    def close(self) -> None:
        pass

dp.create_streaming_table("sensorAlerts")

# Define the schema for the Kafka message value
sensor_schema = StructType([
    StructField("sensor_id", LongType(), False),
    StructField("sensor_type", StringType(), False),
    StructField("sensor_value", LongType(), False)])

@dp.append_flow(target = "sensorAlerts")
def kafka_delta_flow():
    return (
      spark.readStream
        .format("kafka")
        .option("subscribe", KAFKA_TOPIC)
        .option("startingOffsets", "earliest")
        .load()
        .select(from_json(col("value").cast("string"), sensor_schema).alias("data"), col("timestamp"))
        .select("data.*", "timestamp")
        .withWatermark('timestamp', '1 hour')
        .groupBy(col("sensor_id"))
        .transformWithStateInPandas(
          statefulProcessor = SensorHeartbeatProcessor(),
          outputStructType = output_schema,
          outputMode = 'update',
          timeMode = 'ProcessingTime'))

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-07-10

Использовать потоки в конвейерах Lakeflow

Пример. Создание потока по умолчанию

SQL

Питон

Пример. Определение потока отдельно от целевого объекта

Питон

SQL

Пример. Запись в потоковую таблицу из нескольких разделов Kafka

Питон

SQL

Пример: Запуск одноразового заполнения данных

Питон

SQL

Пример: Используйте обработку потока добавления вместо UNION

Питон

SQL

Пример. Использование transformWithState для мониторинга пульса датчика

Обратная связь

Дополнительные ресурсы

Пример: Используйте обработку потока добавления вместо `UNION`

Пример. Использование `transformWithState` для мониторинга пульса датчика