foreach_sink

Le @dp.foreach_sink() décorateur définit un ForEach puits qui traite un flux ligne par ligne avec une logique Python personnalisée. Vous référencez le puits comme un target flux d’append ou un flux de mise à jour pour écrire les données. Pour des conseils conceptuels, des considérations et des exemples, voir Utiliser un ForEach puits pour écrire chaque ligne dans les pipelines. Pour traiter chaque micro-lot en tant que DataFrame, voir foreach_batch_sink.

Syntax

Pour définir un ForEach puits à partir d’une fonction, décorez la fonction avec @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.

Pour définir un ForEach puits à partir d’un objet écrivain, appelez dp.foreach_sink() et passez l’objet avec l’argument writer :

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

Vous pouvez aussi passer une fonction à dp.foreach_sink() avec l’argument func au lieu d’utiliser le décorateur, par dp.foreach_sink(name="<name>", func=row_handler)exemple .

Parameters

Parameter	Description
ligne	Required. Un Row document qui contient un enregistrement du stream. Votre travail, ou la process méthode de votre rédacteur, reçoit une ligne par appel.
name	Optional. Nom unique permettant d’identifier le récepteur dans le pipeline. Par défaut, le nom de la fonction est utilisé. Pour un objet écrivain, par défaut le nom de la classe est écrit en minuscules.
func	Optional. La fonction que le pipeline appelle pour chaque ligne. Utilisez-les func quand vous appelez dp.foreach_sink() directement au lieu d’être décorateur. Vous ne pouvez pas réussir les deux func et writer.
writer	Optional. Un objet qui implémente la process(row) méthode, et éventuellement les open(partition_id, epoch_id) méthodes et close(error) . Présentez-le comme un argument de mot-clé. Vous ne pouvez pas réussir les deux func et writer.

Méthodes d’écriture

Un objet écrivain utilise les mêmes méthodes que l’écrivain Apache Spark Structured Streaming foreach . Pour chaque partition de chaque micro-batch, le pipeline appelle open, puis appelle process pour chaque ligne si open retourne True, puis appelle close. Pour des informations sur l’écriture de sorties personnalisées en streaming structuré, voir Utiliser foreachBatch pour écrire sur des puits de données arbitraires.

Méthode	Description
open(partition_id, epoch_id)	Optional. Le pipeline appelle cette méthode avant de traiter les lignes d’une partition pour un micro-batch. Utilisez-le pour ouvrir des connexions ou lancer des transactions. Revenez True pour traiter les lignes de la partition, ou False pour les sauter. Si l’auteur ne définit openpas , le pipeline traite les lignes de chaque partition. partition_id et epoch_id peuvent répéter entre les flux qui écrivent dans le même grand nombre et après un rafraîchissement complet. Avant de les utiliser pour dédupliquer des lignes, voir Gérer les échecs et dupliquer les lignes.
process(row)	Required. Le pipeline appelle cette méthode pour chaque ligne.
close(error)	Optional. Le pipeline appelle cette méthode après avoir traité les lignes de la partition. error est l’exception survenant lors du traitement des lignes, ou None si le traitement réussissait. Utilisez-le pour rapprocher les contacts. Si open une transaction a été lancée, commencez seulement lorsque error est None, et revenez sinon.