foreach_sink

O @dp.foreach_sink() decorador define um ForEach sumidouro, que processa um fluxo uma linha de cada vez com lógica Python personalizada. Refere-se ao sumidouro como um target fluxo de adição ou de atualização para escrever os dados. Para orientações conceptuais, 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())

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
linha	Required. A Row que contém um registo do stream. A sua função, ou o process método do seu escritor, recebe uma linha por cada chamada.
name	Optional. Um nome único para identificar o sumidouro dentro do oleoduto. Por defeito, o nome da função. Para um objeto escritor, o nome da classe é definido em minúscula.
func	Optional. A função que o pipeline chama para cada linha. Use func quando ligar dp.foreach_sink() diretamente em vez de como decorador. Não podes 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) Apresenta-o como um argumento de palavra-chave. Não podes passar em ambos func e writer.

Métodos de escrita

Um objeto escritor utiliza os mesmos métodos que o escritor Apache Spark Structured Streaming foreach . Para cada partição de cada micro-lote, o pipeline chama open, depois chama process para cada linha se open retorna True, e depois chama close. Para informações sobre a escrita de saídas personalizadas em Structured Streaming, veja Use foreachBatch para escrever em dispersões de dados arbitrárias.

Method	Description
open(partition_id, epoch_id)	Optional. O pipeline chama este método antes de processar as linhas de uma partição para um micro-batch. Use-o para abrir ligações ou iniciar transações. Voltar True para processar as linhas da partição, ou False para as ignorar. Se o autor não definir open, o pipeline processa as linhas de cada partição. partition_id e epoch_id pode repetir-se entre fluxos que escrevem no mesmo sumidouro e após uma atualização completa. Antes de os usar para desduplicar linhas, veja Tratar falhas e duplicar linhas.
process(row)	Required. O pipeline chama este método para cada linha.
close(error)	Optional. O pipeline chama este método depois de 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-o para fechar ligações. Se open iniciar uma transação, compromete-a apenas quando error for None, e reverte-a caso contrário.