Utilize um ForEach destino para escrever cada linha nos pipelines

Utilize um ForEach sink para escrever dados em streaming para destinos sem sink incorporado, como uma API REST externa ou um armazenamento de chave-valor.

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

Quando usar um ForEach sumidouro

Utilize um ForEach sink quando precisar de escrever cada registo individualmente num destino que não tenha formato de sink incorporado, como chamar uma API externa ou atualizar um armazenamento chave-valor para cada registo. Use um ForEachBatch sink em vez disso quando a sua lógica funciona num micro-batch completo, como fundir numa tabela Delta Lake ou escrever para vários destinos com operações DataFrame.

Esta tabela compara os dois destinos:

Para informações sobre destinos incorporados, consulte Destinos em pipelines do Lakeflow. Para a @dp.foreach_sink() referência da API Python, consulte foreach_sink.

Defina um ForEach destino

Podes definir um ForEach sink a partir de uma função ou de um objeto escritor.

Para definir o destino a partir de uma função, decore-a com @dp.foreach_sink(). O pipeline chama a função para cada linha. Se omitir name, o sink usa o nome da função:

from pyspark import pipelines as dp

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

Para definir o sumidouro a partir de um objeto de escrita, passe o objeto com o argumento nomeado writer. O escritor deve implementar process(row) e pode implementar open(partition_id, epoch_id) e close(error). Usa um writer quando quiseres abrir uma ligação para cada partição e reutiliza-o para cada linha. Se omitir name, o sink usa o nome da classe do writer, em minúsculas:

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 detalhes sobre cada método, consulte métodos de Writer.

Entradas do registo de eventos

Quando defines um ForEach sink, o log de eventos do pipeline regista um SinkDefinition evento com "format": "foreach" para o sink.

Escrever para um ForEach sink

Podes escrever para um ForEach destino a partir de um fluxo de adição ou de um fluxo de atualização.

Escreva com um fluxo de acréscimo

Para escrever para um ForEach sink a partir de um fluxo de acréscimo, faça referência ao nome do sink como target:

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

Podes definir múltiplos fluxos de anexação que visam o mesmo ForEach destino. Cada fluxo mantém o seu próprio ponto de controlo.

Escrever com um fluxo atualizado

Para escrever num ForEach sink a partir de um fluxo de atualização, use o @dp.update_flow decorador:

@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 chave para o sink novamente num micro-lote posterior com um novo valor. Para substituir o valor antigo, faz upsert de cada linha no teu destino por chave em vez de a inserir. No exemplo anterior, use user_id como chave para que cada nova contagem para um utilizador substitua a contagem anterior.

Para processamento de baixa latência, um fluxo de atualização que escreve num ForEach sink pode ser executado em modo em tempo real. Veja Usar o modo em tempo real nos pipelines Lakeflow.

Importar bibliotecas e dependências nos executores

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

• Adicione os pacotes que o seu código precisa ao ambiente do pipeline. Pacotes no ambiente de pipeline estão disponíveis para o seu código nos executores. Consulte Gerenciar dependências Python para pipelines.
• Importa as bibliotecas que o teu código usa dentro da função ou dos métodos de escritor, não apenas no topo do ficheiro fonte do pipeline.
• Não dependa dos objetos que cria no driver, como um cliente ou conexão que abre no topo do ficheiro. Crie-as dentro da função, ou no método open do gravador.

Lidar com falhas e linhas duplicadas

Se o seu código levantar uma exceção, o fluxo falha e o registo de eventos do pipeline regista um evento de erro que inclui a mensagem de exceção.

Quando um micro-batch é tentado novamente após uma falha, o sink pode receber algumas linhas novamente. Torna as tuas escritas por linha idempotentes, por exemplo, usando uma chave única da linha para upsert no destino. Ver Garantias de processamento em pipelines do Lakeflow.

Se usar um objeto escritor, open recebe um partition_id e um epoch_id. Juntos, estes IDs identificam as linhas de uma partição num micro-batch, para que possa usá-los para desduplicar linhas. Uma função não recebe estes IDs.

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

• Vários fluxos escrevem para o mesmo sink.
• Um recarregamento completo redefine os pontos de controlo.

Para desduplicar com estes IDs, escreve no sink a partir de um único fluxo e limpa o estado de deduplicação quando executares uma atualização completa.

Atualização completa

Numa atualização completa, o pipeline reinicia o checkpoint de cada fluxo que escreve no ForEach destino. O pipeline não rastreia nem limpa os dados que o sink escreve, por isso geres a retenção e a limpeza para o teu destino. Se precisares de recomeçar do zero, remove esses dados tu próprio.

Limitations

• Métricas de linhas: O pipeline não reporta o número de linhas que um ForEach sink escreve.
• Chamadas por linha: Se o seu código chamar um sistema externo para cada linha, o sistema externo pode limitar o débito. Para reduzir a sobrecarga, abra as ligações pelo open método do escritor e reutilize-as para cada linha, ou use um ForEachBatch sink para escrever em grande quantidade.

Examples

Estes exemplos mostram formas comuns de usar um ForEach sink.

Exemplo de sintaxe básica

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

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

Escrever cada linha numa API externa

Este exemplo utiliza um objeto writer para enviar cada ordem para uma API REST externa. O escritor abre uma sessão HTTP para cada partição. Utiliza o ID da encomenda na URL, por isso uma linha repetida atualiza o mesmo registo em vez de criar um duplicado:

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

Escrever para um sink a partir de múltiplos fluxos

Este exemplo escreve duas fontes para o mesmo ForEach sink:

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

Recursos adicionais

• foreach_sink
• Use ForEachBatch para escrever em destinos de dados arbitrários em pipelines
• Use foreachBatch para escrever em destinos de dados arbitrários
• Garantias de processamento em oleodutos Lakeflow
• Utilizar o modo em tempo real nos pipelines Lakeflow