append_flow

De @dp.append_flow decorator maakt toevoegstromen of backfills voor uw pijplijntabellen. De functie moet een Apache Spark-streaming dataframe retourneren. Zie Incrementeel gegevens laden en verwerken met Lakeflow-pijplijnstromen.

Append-flows kunnen zich richten op streamingtabellen, beheerde tabellen of sinks.

Syntaxis

from pyspark import pipelines as dp

dp.create_streaming_table("<target-table-name>") # Required only if the target table doesn't exist.

@dp.append_flow(
  target = "<target-table-name>",
  name = "<flow-name>", # optional, defaults to function name
  once = False, # optional
  depends_on = "<flow-name>", # optional, Public Preview
  spark_conf = {"<key>" : "<value", "<key" : "<value>"}, # optional
  comment = "<comment>", # optional
  import_checkpoint = "<checkpoint-path>") # optional
def <function-name>():
  return (<streaming-query>) #

Parameterwaarden

Kenmerk	Typologie	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 tabel of sink die het doel is van de toevoegstroom.
name	str	De naam van de stroom. Als deze niet is opgegeven, wordt standaard de functienaam gebruikt.
once	bool	U kunt de stroom desgewenst definiëren als een eenmalige stroom, zoals een backfill. Door once=True te gebruiken verandert de stroom op twee manieren: De retourwaarde. streaming-query. moet in dit geval een Batch DataFrame zijn, niet een streaming DataFrame. De stroom wordt standaard één keer uitgevoerd. Als de pijplijn wordt bijgewerkt met een volledige vernieuwing, wordt de ONCE stroom opnieuw uitgevoerd om de gegevens opnieuw te maken.
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 lijst met Spark-configuraties voor de uitvoering van deze query
import_checkpoint	str	Het pad naar een bestaand Structured Streaming-checkpoint om in de flow te importeren, zodat een gemigreerde stream wordt hervat vanaf zijn laatst toegewezen offset in plaats van de bron opnieuw te verwerken. Het importeren van een checkpoint is in de bèta. Zie Migrate a Structured Streaming checkpoint.

Voorbeelden

from pyspark import pipelines as dp

# Create a sink for an external Delta table
dp.create_sink("my_sink", "delta", {"path": "/tmp/delta_sink"})

# Add an append flow to an external Delta table
@dp.append_flow(name = "flow", target = "my_sink")
def flowFunc():
  return <streaming-query>

# Add a backfill
@dp.append_flow(name = "backfill", target = "my_sink", once = True)
def backfillFlowFunc():
    return (
      spark.read
      .format("json")
      .load("/path/to/backfill/")
    )

# Create a Kafka sink
dp.create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

# Add an append flow to a Kafka sink
@dp.append_flow(name = "flow", target = "my_kafka_sink")
def myFlow():
  return read_stream("xxx").select(F.to_json(F.struct("*")).alias("value"))

Migreer een gestructureerd streaming-controlepunt

Important

Het importeren van een checkpoint is in de bèta.

Gebruik import_checkpoint om een bestaande Structured Streaming-workload naar een pipeline te migreren zonder de bron opnieuw te verwerken. Stel het in op de checkpointLocation gebruikte Structured Streaming-query, wat een cloudopslag, Unity Catalog-volume of DBFS-pad kan zijn. Bij de eerste pipeline-update kloont de flow dat checkpoint naar de beheerde opslag van de pipeline. De stroom wordt vervolgens hervat vanaf de laatst toegewezen offset met de staat (zoals aggregaties, deduplicatiesleutels en watermerken) intact. Latere updates van de pijplijn gebruiken het gekloonde checkpoint van de flow; het oorspronkelijke checkpoint wordt niet gewijzigd.

De flow moet gericht zijn op een beheerde tabel die met create_table of een sink is gemaakt.

Stop de oorspronkelijke Structured Streaming-query voordat je de pipeline uitvoert. De oorspronkelijke Structured Streaming-query kan na de import opnieuw worden gebruikt, maar je moet de checkpointstatus beheren en ervoor zorgen dat de pipeline en de query niet tegelijk naar dezelfde tabel schrijven, wat dubbele data kan opleveren.

Maak de Structured Streaming-query opnieuw aan als een pijplijnflow die naar een nieuwe tabel schrijft en het checkpoint importeert:

from pyspark import pipelines as dp

# Create a new managed table for the pipeline
dp.create_table("target_table")

# Continue from the imported checkpoint instead of reprocessing the source.
@dp.append_flow(
  target = "target_table",
  import_checkpoint = "/Volumes/my_catalog/my_schema/checkpoints/my_stream",
)
def migrate():
  # The same source your original query read from.
  return spark.readStream.table("source_table")

Het checkpoint wordt slechts één keer geïmporteerd, bij de eerste pipeline-update; latere updates negeren import_checkpoint. Een volledige verversing importeert het checkpoint niet opnieuw; het begint vanaf een nieuw, leeg checkpoint en verwerkt de bron opnieuw. Om een ander controlepunt te importeren, gebruik je een flownaam die nog niet eerder voor de doeltabel is gebruikt; het hergebruiken van een bestaande flownaam slaat de import over.

Limitations

• Het importeren van een checkpoint in een bestaande tabel (bijvoorbeeld het oorspronkelijke Structured Streaming-querydoel) wordt niet ondersteund. Richt je op een nieuwe tabel die de pipeline aanmaakt, of een sink.
• import_checkpoint wordt alleen ondersteund op append_flow.