append_flow

Dekoratören @dp.append_flow skapar tilläggsflöden eller påfyllningar för dina pipelinetabeller. Funktionen måste returnera en Apache Spark streaming-Dataram. Se Läsa in och bearbeta data stegvis med Lakeflow-pipelineflöden.

Appendflöden kan rikta sig mot strömmande tabeller, hanterade tabeller eller sinkar.

Syntax

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

Parameterar

Parameter Typ Description
function function Obligatoriskt. En funktion som returnerar en Apache Spark-strömmande DataFrame från en användardefinierad fråga.
target str Obligatoriskt. Namnet på den tabell eller mottagare som är målet för tilläggsflödet.
name str Flödets namn Om det inte anges används funktionsnamnet som standard.
once bool Du kan också definiera flödet som ett engångsflöde, till exempel en återfyllnad. Om du använder once=True ändras flödet på två sätt:
  • Returvärdet. streaming-query. måste vara en batchdataram i det här fallet, inte en strömmande dataram.
  • Flödet körs en gång som standard. Om pipelinen uppdateras med en komplett uppdatering, så körs flödet igen för att återskapa data.
depends_on str eller list Offentlig förhandsversion. Ett eller flera flödesnamn som måste slutföras framgångsrikt innan detta flöde startar. Accepterar ett enda flödesnamn eller en lista med namn. Detta beordrar endast flödesexekvering; det förändrar inte hur flödet körs. Se Order pipeline flow execution with depends_on.
comment str En beskrivning av flödet.
spark_conf dict En lista över Spark-konfigurationer för körning av den här frågan
import_checkpoint str Vägen till en befintlig Structured Streaming-kontrollpunkt importeras till flödet, så att en migrerad ström återupptas från sin senaste committerade offset istället för att återbearbeta källan. Att importera en kontrollpunkt är i Beta. Se Migrera en strukturerad streamingkontrollpunkt.

Examples

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

Migrera en strukturerad strömningskontroll

Important

Att importera en kontrollpunkt är i Beta.

Använd import_checkpoint för att migrera en befintlig Structured Streaming-arbetsbelastning till en pipeline utan att behöva återbearbeta källkoden. Ställ in den som checkpointLocation din Structured Streaming-fråga används, vilket kan vara en molnlagring, en Unity Catalog-volym eller en DBFS-sökväg. Vid den första pipelineuppdateringen klonar flödet den checkpointen in i pipelinens hanterade lagring. Flödet återupptas sedan från den senaste committerade offset med dess tillstånd (såsom aggregeringar, dedupliceringsnycklar och vattenstämplar) intakt. Efterföljande pipelineuppdateringar använder flödets klonade kontrollpunkt; den ursprungliga kontrollpunkten modifieras inte.

Flödet måste rikta in sig på en hanterad tabell skapad med create_table eller en sink.

Stoppa den ursprungliga Structured Streaming-frågan innan du kör pipelinen. Den ursprungliga Structured Streaming-frågan kan återanvändas efter importen, men du måste hantera dess kontrollpunkt och se till att pipelinen och frågan inte skriver till samma tabell samtidigt, vilket kan ge dubbletter av data.

Återskapa Structured Streaming-frågan som ett pipelineflöde som skriver till en ny tabell och importerar dess kontrollpunkt:

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

Kontrollpunkten importeras endast en gång, vid den första pipelineuppdateringen; senare uppdateringar ignorerar import_checkpoint. En fullständig uppdatering importerar inte kontrollpunkten igen; den startar från en ny, tom kontrollpunkt och bearbetar källkoden på nytt. För att importera en annan kontrollpunkt, använd ett flödesnamn som inte har använts för måltabellen tidigare; att återanvända ett befintligt flödesnamn hoppar över importen.

Limitations

  • Att importera en kontrollpunkt till en tabell som redan finns (till exempel det ursprungliga Structured Streaming-frågemålet) stöds inte. Sikta på en ny tabell som pipelinen skapar, eller en sänka.
  • import_checkpoint stöds endast på append_flow.
  • Uppdaterades senast