foreach_sink

O decorador @dp.foreach_sink() define um sink ForEach, que processa uma transmissão uma linha de cada vez com lógica Python personalizada. Você faz referência ao sink como um target em um fluxo de adição ou um fluxo de atualização para gravar os dados. Para obter orientação conceitual, considerações e exemplos, consulte Usar um sink ForEach para gravar cada linha em pipelines. Para processar cada micro-lote como um DataFrame, consulte foreach_batch_sink.

Sintaxe

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

Python

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 sink ForEach de um objeto de gravador, chame dp.foreach_sink() e passe o objeto com o argumento writer:

Python

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 argumento func em vez de usar o decorador, por exemplo dp.foreach_sink(name="<name>", func=row_handler).

Parâmetros

Parâmetro	Descrição
row	Obrigatório. Uma Row que contém um registro da transmissão. Sua função, ou o método process do seu writer, recebe uma linha para cada chamada.
name	Opcional. Um nome exclusivo para identificar o sink dentro do pipeline. O default é o nome da função se não for especificado. Para um objeto writer, o default é o nome da classe em minúsculas.
func	Opcional. A função que o pipeline chama para cada linha. Use func quando você chamar dp.foreach_sink() diretamente em vez de como um decorador. Você não pode passar func e writer ao mesmo tempo.
writer	Opcional. Um objeto que implementa o método process(row) e, opcionalmente, os métodos open(partition_id, epoch_id) e close(error). Passe-o como um argumento de palavra-chave. Você não pode passar func e writer ao mesmo tempo.

Parâmetro	Descrição
row	Obrigatório. Uma Row que contém um registro da transmissão. Sua função, ou o método process do seu writer, recebe uma linha para cada chamada.
name	Opcional. Um nome exclusivo para identificar o sink dentro do pipeline. O default é o nome da função se não for especificado. Para um objeto writer, o default é o nome da classe em minúsculas.
func	Opcional. A função que o pipeline chama para cada linha. Use func quando você chamar dp.foreach_sink() diretamente em vez de como um decorador. Você não pode passar func e writer ao mesmo tempo.
writer	Opcional. Um objeto que implementa o método process(row) e, opcionalmente, os métodos open(partition_id, epoch_id) e close(error). Passe-o como um argumento de palavra-chave. Você não pode passar func e writer ao mesmo tempo.

Métodos de writer

Um objeto escritor usa os mesmos métodos que o escritor foreach do Structured Streaming do Apache Spark. Para cada partição de cada micro-lote, o pipeline chama open, então chama process para cada linha se open retornar True, e então chama close. Para obter informações sobre como gravar saídas personalizadas no Structured Streaming, consulte Usar foreachBatch para gravar em coletores de dados arbitrários.

Método	Descrição
open(partition_id, epoch_id)	Opcional. O pipeline chama esse método antes de processar as linhas de uma partição para um micro-lotes. Use-o para abrir conexões ou começar transações. Retorne True para processar as linhas da partição, ou False para ignorá-las. Se o gravador não definir open, o pipeline processa as linhas de cada partição. partition_id e epoch_id podem se repetir em fluxos que gravam no mesmo coletor e após um refresh completo. Antes de usá-los para deduplicar linhas, consulte Manipular falhas e linhas duplicadas.
process(row)	Obrigatório. O pipeline chama esse método para cada linha.
close(error)	Opcional. O pipeline chama esse método depois de processar as linhas da partição. error é a exceção que ocorreu durante o processamento das linhas, ou None se o processamento teve êxito. Use-o para fechar conexões. Se open começou uma transação, faça commit dela apenas quando error for None e faça rollback dela caso contrário.

Método	Descrição
open(partition_id, epoch_id)	Opcional. O pipeline chama esse método antes de processar as linhas de uma partição para um micro-lotes. Use-o para abrir conexões ou começar transações. Retorne True para processar as linhas da partição, ou False para ignorá-las. Se o gravador não definir open, o pipeline processa as linhas de cada partição. partition_id e epoch_id podem se repetir em fluxos que gravam no mesmo coletor e após um refresh completo. Antes de usá-los para deduplicar linhas, consulte Manipular falhas e linhas duplicadas.
process(row)	Obrigatório. O pipeline chama esse método para cada linha.
close(error)	Opcional. O pipeline chama esse método depois de processar as linhas da partição. error é a exceção que ocorreu durante o processamento das linhas, ou None se o processamento teve êxito. Use-o para fechar conexões. Se open começou uma transação, faça commit dela apenas quando error for None e faça rollback dela caso contrário.