foreach_sink

O @dp.foreach_sink() decorador define um ForEach sumidouro, que processa um fluxo linha por linha com lógica Python personalizada. Você faz referência ao sumidouro como um target fluxo de anexo ou um fluxo de atualização para gravar os dados. Para orientações conceituais, considerações e exemplos, veja Use a ForEach sink para escrever cada linha nos pipelines. Para processar cada micro-lote como um DataFrame, veja foreach_batch_sink.

Syntax

Para definir um ForEach sumidouro a partir de uma função, decore a função com @dp.foreach_sink():

from pyspark import pipelines as dp

@dp.foreach_sink(name="<name>")
def row_handler(row):
    """
    Required:
      - `row`: a Row that contains one record of the stream.
    """
    # Your custom write logic here.
    # This function runs on the executors, so import any libraries it uses inside the function.

Para definir um ForEach sumidouro a partir de um objeto escritor, chame dp.foreach_sink() e passe o objeto com o writer argumento:

from pyspark import pipelines as dp

class RowWriter:
    def open(self, partition_id, epoch_id):
        # Optional. Return False to skip the rows of this partition.
        return True

    def process(self, row):
        # Required. Called for each row.
        ...

    def close(self, error):
        # Optional. `error` is None if processing succeeded.
        ...

dp.foreach_sink(name="<name>", writer=RowWriter())

Você também pode passar uma função para dp.foreach_sink() com o func argumento em vez de usar o decorador, por exemplo dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parameter	Description
row	Required. A Row que contém um registro do stream. Sua função, ou o process método do seu escritor, recebe uma linha para cada chamada.
name	Optional. Um nome exclusivo para identificar o coletor dentro do canal de processamento. O nome padrão é a função. Para um objeto escritor, o nome da classe é usado por padrão em minúsculas.
func	Optional. A função que o pipeline chama para cada linha. Use func quando você ligar dp.foreach_sink() diretamente em vez de como decorador. Você não pode passar em ambos func e writer.
writer	Optional. Um objeto que implementa o process(row) método e, opcionalmente, os open(partition_id, epoch_id) métodos e.close(error) Passe como um argumento de palavra-chave. Você não pode passar em ambos func e writer.

Métodos de escritor

Um objeto escritor usa os mesmos métodos do escritor Apache Spark Structured Stream foreach . Para cada partição de cada micro-lote, o pipeline chama open, então chama process para cada linha se open retorna True, e então chama close. Para informações sobre como escrever saídas personalizadas em Structured Streaming, veja Use foreachBatch para escrever em sumidouros de dados arbitrários.

Método	Description
open(partition_id, epoch_id)	Optional. O pipeline chama esse método antes de processar as linhas de uma partição para um micro-batch. Use para abrir conexões ou iniciar transações. Volte True para processar as linhas da partição, ou False para pulá-las. Se o escritor não definir open, o pipeline processa as linhas de cada partição. partition_id e epoch_id pode repetir fluxos que gravam no mesmo sumidouro e após uma atualização completa. Antes de usá-las para desduplicar linhas, veja Lidar com falhas e duplicar linhas.
process(row)	Required. O pipeline chama esse método para cada linha.
close(error)	Optional. O pipeline chama esse método após processar as linhas da partição. error é a exceção que ocorreu durante o processamento de linhas, ou None se o processamento foi bem-sucedido. Use para fechar conexões. Se open iniciar uma transação, faça commit apenas quando error for None, e reverta caso contrário.