Use um sink ForEach para gravar cada linha em pipelines

Use um destino ForEach para gravar dados de transmissão em destinos que não possuem um destino integrado, como uma API REST externa ou um armazenamento de key-valor.

O coletor ForEach processa uma transmissão uma linha por vez. Para cada linha, o pipeline chama uma função Python personalizada ou um objeto de gravação, semelhante ao coletor foreach no Apache Spark Structured Streaming.

nota

O sink ForEach descrito nesta página é para pipelines. O Apache Spark Structured Streaming também suporta foreach. Para obter informações sobre como escrever saídas personalizadas no Structured Streaming, consulte Usar o foreachBatch para gravar em sinks de dados arbitrários.

Quando usar um sink ForEach

Use um sink ForEach quando precisar gravar cada registro individualmente em um destino que não tenha um formato de sink integrado, como chamar uma API externa ou atualizar um armazenamento de key-valor para cada registro. Use um ForEachBatch sink em vez disso quando sua lógica funcionar em um micro-lote inteiro, como mesclar em uma tabela Delta Lake ou gravar em vários destinos com operações DataFrame.

Esta tabela compara os dois sinks:

Característica	ForEach sink	ForEachBatch sink
Unidade de processamento	Uma linha de cada vez	Um micro-lote de DataFrame por vez
Onde seu código é execução	Nos executores	No driver
API do Python	@dp.foreach_sink()	@dp.foreach_batch_sink()

Característica	ForEach sink	ForEachBatch sink
Unidade de processamento	Uma linha de cada vez	Um micro-lote de DataFrame por vez
Onde seu código é execução	Nos executores	No driver
API do Python	@dp.foreach_sink()	@dp.foreach_batch_sink()

Para obter informações sobre sinks integrados, consulte Sinks em LakeFlow Pipelines. Para a referência da API Python @dp.foreach_sink(), consulte foreach_sink.

Definir um sink de ForEach

Você pode definir um sink ForEach de uma função ou de um objeto de escrita.

Para definir o sink de uma função, decore a função com @dp.foreach_sink(). O pipeline chama a função para cada linha. Se você omitir name, o sink usará o nome da função:

Python

from pyspark import pipelines as dp

@dp.foreach_sink()
def my_row_sink(row):
    # Custom per-row logic here.
    ...

Para definir o sink de um objeto writer, passe o objeto com o argumento de palavra-chave writer. O writer deve implementar process(row) e pode implementar open(partition_id, epoch_id) e close(error). Use um writer quando quiser abrir uma conexão para cada partição e reutilizá-la para cada linha. Se você omitir name, o sink usa o nome da classe do writer, em minúsculas:

Python

from pyspark import pipelines as dp

class MyRowWriter:
    def open(self, partition_id, epoch_id):
        # Open a connection for this partition.
        return True

    def process(self, row):
        # Write the row by using the connection.
        ...

    def close(self, error):
        # Close the connection.
        ...

dp.foreach_sink(name="my_row_sink", writer=MyRowWriter())

Para obter detalhes sobre cada método, consulte Métodos de gravação.

Entradas do log de eventos

Quando você define um sink ForEach, o log de eventos do pipeline registra um evento SinkDefinition com "format": "foreach" para o sink.

Gravar em um coletor de ForEach

Você pode gravar em um sink ForEach a partir de um fluxo de append ou de um fluxo de update.

Gravar com um fluxo de anexação

Para gravar em um coletor de ForEach de um fluxo de anexo, faça referência ao nome do coletor como o target:

Python

@dp.append_flow(target="my_row_sink")
def my_flow():
    return spark.readStream.table("my_catalog.my_schema.source_table")

Você pode definir múltiplos fluxos de adição que visam o mesmo sink ForEach. Cada fluxo mantém seu próprio ponto de verificação.

Escreva com um fluxo de atualização

Para gravar em um sink ForEach de um fluxo de atualização, use o decorador @dp.update_flow:

Python

@dp.update_flow(target="my_row_sink")
def my_stateful_flow():
    return spark.readStream.table("my_catalog.my_schema.events").groupBy("user_id").count()

Um fluxo de atualização pode enviar a mesma key para o sink novamente em um micro-lote posterior com um novo valor. Para substituir o valor antigo, upsert cada linha em seu destino por key em vez de inseri-la. No exemplo anterior, use user_id como a key, de modo que cada nova contagem para um usuário substitua a contagem anterior.

Para processamento de baixa latência, um fluxo de atualização que grava em um sink ForEach pode ser execução em modo de tempo real. Consulte Usar modo em tempo real em LakeFlow Pipelines.

Importe bibliotecas e dependências nos executores

A sua função ou escritor é executado nos executores, não no driver. Siga estas práticas:

• Adicione os pacotes de que seu código precisa ao ambiente do pipeline. Os pacotes no ambiente do pipeline estão disponíveis para seu código nos executores. Consulte Gerenciar dependências Python para pipelines.
• Importe as bibliotecas que seu código usa dentro da função ou dos métodos de escrita, não apenas no topo do arquivo de origem do pipeline.
• Não dependa de objetos que você criou no driver, como um cliente ou conexão que você abre no topo do arquivo. Crie-os dentro da função ou no método open do writer.

Lidar com falhas e linhas duplicadas

Se o seu código gerar uma exceção, o fluxo falhará e o log de eventos do pipeline registrará um evento de erro que incluirá a mensagem de exceção.

Quando um micro-lote é tentado novamente após uma falha, o sink pode receber algumas linhas novamente. Torne suas gravações por linha idempotentes, por exemplo, usando uma key exclusiva da linha para fazer upsert no destino. Consulte Garantias de processamento em LakeFlow Pipelines.

Se você usar um objeto writer, open receberá um partition_id e um epoch_id. Juntos, esses IDs identificam as linhas de uma partição em um micro-lote, para que você possa usá-los para remover duplicatas de linhas. Uma função não recebe esses IDs.

Os IDs são exclusivos apenas dentro do checkpoint de um fluxo. O mesmo par pode se repetir nos seguintes casos:

• Vários fluxos escrevem no mesmo sink.
• Um refresh completo Reset os checkpoints.

Para desduplicar com esses IDs, escreva para o sink a partir de um único fluxo e limpe seu estado de desduplicação quando você fizer uma execução de um refresh completo.

refresh completo

Em um refresh completo, o pipeline faz o reset do checkpoint de cada fluxo que grava no sink ForEach. O pipeline não rastreia nem limpa os dados que o sink grava, portanto, você gerencia a retenção e a limpeza para seu destino. Se você precisa de um começar limpo, remova esses dados você mesmo.

Limitações

• Métricas de linha : o pipeline não relata o número de linhas que um sink ForEach grava.
• Chamadas por linha : Se o seu código chamar um sistema externo para cada linha, o sistema externo poderá limitar o throughput. Para reduzir o overhead, abra conexões no método open do writer e reutilize-as para cada linha, ou use um sink ForEachBatch para escrever em massa.

Exemplos

Estes exemplos mostram formas comuns de utilizar um sink ForEach.

Exemplo de sintaxe básica

Este exemplo define um sink ForEach a partir de uma função e escreve nele a partir de um fluxo de acréscimo:

Python

from pyspark import pipelines as dp

# Create a ForEach sink
@dp.foreach_sink(name="my_foreach_sink")
def handle_row(row):
    # Custom per-row logic here.
    ...

# Create source data for the example
@dp.table()
def example_source_data():
    return spark.range(5)

# Write to the sink from an append flow
@dp.append_flow(target="my_foreach_sink")
def my_flow():
    return spark.readStream.table("example_source_data")

Gravar cada linha em uma API externa

Este exemplo usa um objeto escritor para enviar cada pedido para uma API REST externa. O escritor abre uma sessão HTTP para cada partição. Ele usa o ID do pedido na URL, portanto, uma linha repetida atualiza o mesmo registro em vez de criar um duplicado:

Python

from pyspark import pipelines as dp

class OrdersApiWriter:
    def open(self, partition_id, epoch_id):
        import requests  # Import on the executor

        self.session = requests.Session()
        return True

    def process(self, row):
        response = self.session.put(
            f"https://api.example.com/orders/{row.order_id}",
            json={"order_id": row.order_id, "status": row.status},
            timeout=10,
        )
        # Raise on HTTP errors so that the flow fails instead of dropping the row.
        response.raise_for_status()

    def close(self, error):
        self.session.close()

dp.foreach_sink(name="orders_api_sink", writer=OrdersApiWriter())

@dp.append_flow(target="orders_api_sink")
def orders_to_api():
    return spark.readStream.table("my_catalog.my_schema.orders")

Gravar em um sink a partir de vários fluxos

Este exemplo grava duas fontes no mesmo destino ForEach:

Python

from pyspark import pipelines as dp

@dp.foreach_sink(name="events_sink")
def handle_event(row):
    # Custom per-row logic here.
    ...

@dp.append_flow(target="events_sink")
def web_events():
    return spark.readStream.table("my_catalog.my_schema.web_events")

@dp.append_flow(target="events_sink")
def mobile_events():
    return spark.readStream.table("my_catalog.my_schema.mobile_events")

Outros recursos

• foreach_sink
• Usar o ForEachBatch para gravar em coletores de dados arbitrários em pipelines
• Usar foreachBatch para gravar em coletores de dados arbitrários
• Garantias de processamento em LakeFlow Pipelines
• Use o modo em tempo real nos LakeFlow Pipelines