update_flow

Użyj dekoratora @dp.update_flow , aby utworzyć przepływ aktualizacji. Aktualizuj przepływy zapisu w ujściach przy użyciu trybu aktualizacji danych wyjściowych, emitując tylko wiersze, które uległy zmianie w każdej partii. W przeciwieństwie do przepływów dołączania obsługują agregacje stanowe bez konieczności używania znaku wodnego.

Przepływy aktualizacji mogą kierować tylko ujścia. Tabele różnicowe nie są obsługiwane.

Przepływy aktualizacji mogą działać w trybie czasu rzeczywistego dla obciążeń operacyjnych w czasie poniżej sekundy. Tryb czasu rzeczywistego w potokach Lakeflow obsługuje obliczenia klasyczne i serwerowe.

Syntax

from pyspark import pipelines as dp

dp.create_sink("<sink-name>", "<format>", {"<key>": "<value>"})

@dp.update_flow(
  target = "<sink-name>",
  name = "<flow-name>", # optional, defaults to function name
  depends_on = "<flow-name>", # optional, Public Preview
  spark_conf = {"<key>" : "<value>", "<key>" : "<value>"}, # optional
  comment = "<comment>") # optional
def <function-name>():
  return (<streaming-query>)

Parameters

Parameter Typ Description
funkcja function Required. Funkcja, która zwraca strumieniowy DataFrame w Apache Spark na podstawie zapytania zdefiniowanego przez użytkownika.
target str Required. Nazwa ujścia, do których jest zapisywany ten przepływ.
name str Nazwa przepływu. Jeśli nie zostanie podana, wartość domyślna to nazwa funkcji.
depends_on str lub list Publiczna wersja zapoznawcza. Jedna lub więcej nazw przepływów, które muszą zakończyć się pomyślnie, zanim ten przepływ się rozpocznie. Akceptuje pojedynczą nazwę przepływu lub listę nazw. W ciągłym potoku każdy poprzednik musi być jednorazowymonce () przepływem. To nakazuje tylko wykonywanie przepływu; nie zmienia sposobu działania przepływu. Zobacz Wykonywanie przepływu potoku zamówień z depends_on.
comment str Opis przepływu.
spark_conf dict Dykt konfiguracji platformy Spark na potrzeby wykonywania tego zapytania. Te konfiguracje zastępują konfiguracje ustawione dla miejsca docelowego, potoku lub klastra.

Examples

Agregacja do ujścia platformy Kafka

Zapisz wyniki agregacji stanowej do ujścia platformy Kafka:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

dp.create_sink("event_counts_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="event_counts_flow",
    target="event_counts_sink",
)
def event_counts():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .selectExpr("CAST(key AS STRING) AS event_type")
            .groupBy(col("event_type"))
            .count()
    )

Tryb czasu rzeczywistego

Note

Tryb czasu rzeczywistego wymaga potoku skonfigurowanego do ciągłego wykonywania. Zobacz Użyj trybu czasu rzeczywistego w potokach Lakeflow, aby sprawdzić wymagania wykonawcze i włączyć tryb czasu rzeczywistego przed uruchomieniem tego przykładu.

Użyj spark_conf polecenia , aby skonfigurować przepływ aktualizacji dla trybu czasu rzeczywistego:

from pyspark import pipelines as dp

dp.create_sink("my_kafka_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="my_rtm_flow",
    target="my_kafka_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def my_real_time_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
    )

Ograniczenia

  • Ujścia tabeli delty nie są obsługiwane jako obiekty docelowe dla przepływów aktualizacji.
  • Ostatnia aktualizacja o