update_flow

Gebruik de @dp.update_flow decorator om een updatestroom te maken. Werk stromen schrijven naar sinks bij met behulp van de update-uitvoermodus, waarbij alleen de rijen worden verzonden die in elke batch zijn gewijzigd. In tegenstelling tot toevoegstromen bieden ze ondersteuning voor stateful aggregaties zonder dat hiervoor een watermerk is vereist.

Updatestromen kunnen alleen doelsinks gebruiken. Delta-tabellen worden niet ondersteund.

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
function	function	Verplicht. Een functie die een Streaming DataFrame van Apache Spark retourneert vanuit een door de gebruiker gedefinieerde query.
target	str	Verplicht. De naam van de sink naar deze stroom schrijft.
name	str	De naam van de stroom. Als deze niet is opgegeven, wordt standaard de functienaam gebruikt.
depends_on	str of list	Publieke preview. Een of meer stroomnamen die succesvol moeten worden voltooid voordat deze stroom begint. Accepteert een enkele flownaam of een lijst met namen. Dit beveelt alleen de uitvoering van de flow; het verandert niet hoe de flow verloopt. Zie Order pipeline flow execution met depends_on.
comment	str	Een beschrijving voor het proces.
spark_conf	dict	Een dicteerfunctie voor Spark-configuraties voor de uitvoering van deze query. Deze configuraties overschrijven confs set voor de bestemming, pijplijn of cluster.

Examples

Aggregatie naar een Kafka-sink

Stateful aggregatieresultaten schrijven naar een Kafka-sink:

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

Real-time modus

Important

De realtimemodus bevindt zich in openbare preview.

Gebruik spark_conf dit om een updatestroom te configureren voor de realtime-modus:

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

Limitations

• Delta-tabelsinks worden niet ondersteund als doelen voor updatestromen.