Lakeflow işlem hatlarında akışları kullanma

Lakeflow işlem hattındaki akışlar, verileri bir akış tablosuna veya somutlaştırılmış görünüme aktarır. Aşağıdaki örnekler, varsayılan akışların nasıl tanımlanacağını, bir akışın hedefinden ayrı olarak nasıl tanımlanacağını, birden çok Kafka topic’inden bir akış tablosuna nasıl yazılacağını, tek seferlik geri doldurmanın nasıl çalıştırılacağını ve UNION sorgularının eklemeli akış işleme ile nasıl değiştirileceğini gösterir.

Akışlara genel bakış için bkz. Lakeflow işlem hattı akışlarıyla verileri artımlı olarak yükleme ve işleme.

Örnek: Varsayılan akış oluşturma

İşlem hattı oluşturduğunuzda, genellikle destekleyen sorguyla birlikte bir tablo veya görünüm tanımlarsınız. Örneğin, bu sorgu customers_silver'den okuyarak customers_bronze adlı bir akış tablosu oluşturur. Akış tablosu ve varsayılan akışı tek bir adımda birlikte oluşturulur.

SQL

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

Piton

from pyspark import pipelines as dp

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

Akış tablosu için varsayılan akış, her güncelleştirmeye yeni satırlar ekleyen bir ekleme akışıdır ve hedefle aynı ada sahiptir. Bu işlem hatlarını kullanmanın en yaygın yoludur(tek adımda akış ve hedefi oluşturur) ve verileri almak veya dönüştürmek için bunu kullanabilirsiniz. Akış kavramları hakkında daha fazla bilgi için bkz. Lakeflow işlem hattı akışlarıyla verileri artımlı olarak yükleme ve işleme.

Örnek: Akışı hedefinden ayrı olarak tanımlama

Ayrıca, ayrı olarak tanımladığınız bir tablo için akış da oluşturabilirsiniz. Sonuç, akış tablosu ve akış için aynı adın kullanılması da dahil olmak üzere varsayılan akış oluşturmakla aynıdır:

Piton

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

Bir akışı hedefinden ayrı olarak tanımlamak, aynı hedefe veri ekleyen birden çok akış oluşturmanıza olanak tanır. Python arabirimindeki @dp.append_flow dekoratörünü veya SQL arabirimindeki CREATE FLOW...INSERT INTO yan tümcesini kullanarak şunlar gibi görevler için akışlar ekleyin:

• Tam yenileme gerektirmeden mevcut bir akış tablosuna veri ekleyen akış kaynakları ekleyin. Örneğin, çalıştığınız her bölgeden bölgesel verileri birleştiren bir tablonuz olabilir. Yeni bölgeler kullanıma sunulduktan sonra, tam yenileme gerçekleştirmeden yeni bölge verilerini tabloya ekleyebilirsiniz. Bkz . Örnek: Birden çok Kafka konu başlığından akış tablosuna yazma.
• Eksik geçmiş verileri (geri doldurma) ekleyerek bir akış tablosunu güncelleştirin. INSERT INTO ONCE sözdizimini kullanarak tek seferlik çalışan bir geçmişe dönük doldurma oluşturabilirsiniz. Bkz . Örnek: Tek seferlik veri doldurma ve geçmiş verileri işlem hatlarıyla doldurma.
• Birden çok kaynaktaki verileri birleştirin ve UNION yan tümcesini sorguda kullanmak yerine, tek bir veri akışı tablosuna yazın. Yerine ekleme akışı işlemeyi UNION kullanmak, tam yenileme güncelleştirmesi çalıştırmadan hedef tabloyu artımlı olarak güncelleştirmenizi sağlar. Bkz. Örnek: UNION yerine ekleme akışı işleme kullanın.

Python sorguları için create_streaming_table() işlevini kullanarak bir hedef tablo oluşturun.

Örnek: Birden çok Kafka konu başlığından akış tablosuna yazma

Aşağıdaki örnek, "kafka_target" adlı bir akış tablosu oluşturur ve iki Kafka kanalından bu akış tablosuna veri yazar.

Piton

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 sorgularında kullanılan tablo değerli fonksiyon hakkında read_kafka() daha fazla bilgi edinmek için bkz. SQL dil referansında read_kafka.

Python'da program aracılığıyla tek bir tabloyu hedefleyen birden çok akış oluşturabilirsiniz. Aşağıdaki örnekte Kafka konularının listesi için bu desen gösterilmektedir.

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

Örnek: Tek seferlik veri geri doldurma işlemi yürütme

Var olan bir akış tablosuna veri eklemek için bir sorgu çalıştırmak istiyorsanız kullanın append_flow.

Var olan bir veri kümesini ekledikten sonra birden çok seçeneğiniz vardır:

• Sorgunun, backfill dizinine ulaşırsa yeni verileri eklemesini istiyorsanız sorguyu yerinde bırakın.
• Bunun tek seferlik bir dolum olmasını ve bir daha asla çalıştırılmamasını istiyorsanız, işlem hattını bir kez çalıştırdıktan sonra sorguyu kaldırın.
• Sorgunun bir kez çalıştırılmasını ve yalnızca verilerin tamamen güncellendiği durumlarda yeniden çalıştırılmasını istiyorsanız, ekleme sürecinde parametresini once olarak True ayarlayın. SQL'de kullanın INSERT INTO ONCE.

Aşağıdaki örneklerde geçmiş verileri akış tablosuna eklemek için bir sorgu çalıştırılır:

Piton

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"
  );

Daha ayrıntılı bir örnek için bkz. Geçmiş verileri işlem hatlarıyla doldurma.

Örnek: Ekleme akışı işlemesini UNION yerine kullanın

Yan tümcesi olan UNION bir sorgu kullanmak yerine, birden çok kaynağı birleştirmek ve tek bir akış tablosuna yazmak için ekleme akışı sorgularını kullanabilirsiniz. Kullanıcının akış tablosuna birden çok kaynaktan veri eklemesine olanak tanıyan ekleme akış sorgularını, UNION çalıştırmadan yerine kullanabilirsiniz.

Aşağıdaki Python örneği, birden çok veri kaynağını yan tümcesiyle birleştiren bir UNION sorgu içerir:

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

Aşağıdaki örnekler sorguyu UNION ekleme akışı sorgularıyla değiştirir:

Piton

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"
  );

Örnek: Algılayıcı kalp atışlarını izlemek için transformWithState kullanın

Aşağıdaki örnekte, Kafka'dan okuyan ve sensörlerin düzenli aralıklarla sinyal yaydığını doğrulayan durum bilgisine sahip bir işlemci gösterilmektedir. 5 dakika içinde sinyal alınmazsa, işlemci analiz için hedef Delta tablosuna bir giriş oluşturur.

Özel durum bilgisi olan uygulamalarla ilgili daha fazla bilgi için bkz. Özel durum bilgisi olan bir uygulama oluşturma.

"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'))