fluxo_de_acréscimo

O decorador @dp.append_flow cria fluxos de anexação ou preenchimentos retroativos para as tabelas do seu pipeline. A função deve retornar um DataFrame de streaming do Apache Spark. Consulte Carregue e processe dados incrementalmente com fluxos de pipeline do Lakeflow.

Append flows can target streaming tables, managed tables, or sinks.

Sintaxe

Python

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

Parâmetros

Parâmetro	Tipo	Descrição
função	function	Obrigatório. Uma função que retorna um DataFrame de streaming Apache Spark streaming de uma consulta definida pelo usuário.
target	str	Obrigatório. O nome da tabela ou do coletor que é o destino do fluxo de acréscimo.
name	str	O nome do fluxo. Se não for fornecido, o padrão será o nome da função.
once	bool	Opcionalmente, defina o fluxo como um fluxo único, como um aterro. Usar once=True altera o fluxo de duas maneiras: O valor de retorno. streaming-query. deve ser um lotes DataFrame neste caso, não um transmissão DataFrame. O fluxo é executado uma vez por default. Se o pipeline for atualizado com uma refresh completa, o fluxo ONCE será executado novamente para recriar os dados.
depends_on	str ou list	Pré-lançamento público. Um ou mais nomes de fluxo que devem ser concluídos com êxito antes que este fluxo comece. Aceita um único nome de fluxo ou uma lista de nomes. Isso ordena apenas a execução do fluxo; não altera a forma como o fluxo é executado. Consulte Order pipeline flow execution with depends_on.
comment	str	Uma descrição para o fluxo.
spark_conf	dict	Uma lista de configurações do Spark para a execução desta consulta
import_checkpoint	str	The path to an existing Structured Streaming checkpoint to import into the flow, so a migrated transmissão resumes from its last committed offset instead of reprocessing the source. Importing a checkpoint is in Beta. See Migrate a Structured Streaming checkpoint.

Parâmetro	Tipo	Descrição
função	function	Obrigatório. Uma função que retorna um DataFrame de streaming Apache Spark streaming de uma consulta definida pelo usuário.
target	str	Obrigatório. O nome da tabela ou do coletor que é o destino do fluxo de acréscimo.
name	str	O nome do fluxo. Se não for fornecido, o padrão será o nome da função.
once	bool	Opcionalmente, defina o fluxo como um fluxo único, como um aterro. Usar once=True altera o fluxo de duas maneiras: O valor de retorno. streaming-query. deve ser um lotes DataFrame neste caso, não um transmissão DataFrame. O fluxo é executado uma vez por default. Se o pipeline for atualizado com uma refresh completa, o fluxo ONCE será executado novamente para recriar os dados.
depends_on	str ou list	Pré-lançamento público. Um ou mais nomes de fluxo que devem ser concluídos com êxito antes que este fluxo comece. Aceita um único nome de fluxo ou uma lista de nomes. Isso ordena apenas a execução do fluxo; não altera a forma como o fluxo é executado. Consulte Order pipeline flow execution with depends_on.
comment	str	Uma descrição para o fluxo.
spark_conf	dict	Uma lista de configurações do Spark para a execução desta consulta
import_checkpoint	str	The path to an existing Structured Streaming checkpoint to import into the flow, so a migrated transmissão resumes from its last committed offset instead of reprocessing the source. Importing a checkpoint is in Beta. See Migrate a Structured Streaming checkpoint.

Exemplos

Python

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

Migrar um ponto de verificação do Structured Streaming

info

Beta

A importação de um ponto de verificação está em Beta.

Use import_checkpoint to migrate an existing Structured Streaming workload to a pipeline without reprocessing the source. Set it to the checkpointLocation your Structured Streaming query used, which can be a cloud storage, Unity Catalog volume, or DBFS path. On the first pipeline update, the flow clones that checkpoint into the pipeline's managed storage. The flow then resumes from the last committed offset with its state (such as aggregations, deduplication keys, and watermarks) intact. Subsequent pipeline updates use the flow's cloned checkpoint; the original checkpoint is not modified.

O fluxo deve ter como destino uma tabela gerenciada criada com create_table ou um sink.

Stop the original Structured Streaming query before you run the pipeline. The original Structured Streaming query can be reused after the import, but you need to gerenciar its checkpoint state and make sure the pipeline and the query do not write to the same table at the same time, which can produce duplicate data.

Recrie a query do Structured Streaming como um fluxo de pipeline que grava em uma nova tabela e importa seu ponto de verificação:

Python

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

O ponto de verificação é importado apenas uma vez, na primeira atualização do pipeline; as atualizações posteriores ignoram import_checkpoint. Um refresh completo não reimporta o ponto de verificação; ele começa a partir de um novo ponto de verificação vazio e reprocessa a origem. Para importar um ponto de verificação diferente, use um nome de fluxo que não tenha sido usado antes para a tabela de destino; reutilizar um nome de fluxo existente ignora a importação.

Limitações

• Importing a checkpoint into a table that already exists (for example, the original Structured Streaming query target) is not supported. Target a new table that the pipeline creates, or a sink.
• import_checkpoint é compatível apenas em append_flow.