foreach_sink

Le décorateur @dp.foreach_sink() définit un sink ForEach, qui traite un stream une ligne à la fois avec une logique Python personnalisée. Vous référencez le sink en tant que target dans un append flow ou un update flow pour écrire les données. Pour des conseils conceptuels, des considérations et des exemples, voir Utiliser un sink ForEach pour écrire chaque ligne dans les pipelines. Pour traiter chaque micro-batch comme un DataFrame à la place, voir foreach_batch_sink.

Syntaxe

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

Pour définir un récepteur ForEach à partir d’un objet writer, appelez dp.foreach_sink() et passez l’objet avec l’argument 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())

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

Paramètres

parameter	Description
row	Obligatoire. Un Row qui contient un enregistrement du stream. Votre fonction, ou la méthode process de votre writer, reçoit une ligne pour chaque appel.
Nom	Facultatif. Un nom unique pour identifier le récepteur dans le pipeline. default le nom de la fonction est utilisé. Pour un objet writer, le nom de la classe en minuscules est utilisé default.
fonction	Optionnel. La fonction que le pipeline appelle pour chaque ligne. Utilisez func lorsque vous appelez dp.foreach_sink() directement au lieu d’utiliser un décorateur. Vous ne pouvez pas transmettre à la fois func et writer.
writer	Facultatif. Objet qui implémente la méthode process(row) et, éventuellement, les méthodes open(partition_id, epoch_id) et close(error). Passez-le comme un argument de mot clé. Vous ne pouvez pas passer à la fois func et writer.

parameter	Description
row	Obligatoire. Un Row qui contient un enregistrement du stream. Votre fonction, ou la méthode process de votre writer, reçoit une ligne pour chaque appel.
Nom	Facultatif. Un nom unique pour identifier le récepteur dans le pipeline. default le nom de la fonction est utilisé. Pour un objet writer, le nom de la classe en minuscules est utilisé default.
fonction	Optionnel. La fonction que le pipeline appelle pour chaque ligne. Utilisez func lorsque vous appelez dp.foreach_sink() directement au lieu d’utiliser un décorateur. Vous ne pouvez pas transmettre à la fois func et writer.
writer	Facultatif. Objet qui implémente la méthode process(row) et, éventuellement, les méthodes open(partition_id, epoch_id) et close(error). Passez-le comme un argument de mot clé. Vous ne pouvez pas passer à la fois func et writer.

Méthodes d'écriture

Un objet writer utilise les mêmes méthodes que le writer foreach de Structured Streaming pour Apache Spark. Pour chaque partition de chaque micro-batch, le pipeline appelle open, puis appelle process pour chaque ligne si open retourne True, et appelle ensuite close. Pour plus d'informations sur l'écriture de sorties personnalisées dans Structured Streaming, voir Utiliser foreachBatch pour écrire dans des récepteurs de données arbitraires.

Méthode	Description
open(partition_id, epoch_id)	Facultatif. 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 start des transactions. Retourner True pour traiter les lignes de la partition, ou False pour les ignorer. Si le writer ne définit pas open, le pipeline traite les lignes de chaque partition. partition_id et epoch_id peuvent se répéter dans les flux qui écrivent vers le même récepteur et après un refresh complet. Avant de les utiliser pour dédupliquer les lignes, consultez Gestion des échecs et des lignes en double.
process(row)	Obligatoire. Le pipeline appelle cette méthode pour chaque ligne.
close(error)	Facultatif. Le pipeline appelle cette méthode après avoir traité les lignes de la partition. error est l'exception qui s'est produite lors du traitement des lignes, ou None si le traitement a réussi. Utilisez-la pour fermer les connexions. Si open a démarré une transaction, commit uniquement lorsque error est None, et annulez-la sinon.

Méthode	Description
open(partition_id, epoch_id)	Facultatif. 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 start des transactions. Retourner True pour traiter les lignes de la partition, ou False pour les ignorer. Si le writer ne définit pas open, le pipeline traite les lignes de chaque partition. partition_id et epoch_id peuvent se répéter dans les flux qui écrivent vers le même récepteur et après un refresh complet. Avant de les utiliser pour dédupliquer les lignes, consultez Gestion des échecs et des lignes en double.
process(row)	Obligatoire. Le pipeline appelle cette méthode pour chaque ligne.
close(error)	Facultatif. Le pipeline appelle cette méthode après avoir traité les lignes de la partition. error est l'exception qui s'est produite lors du traitement des lignes, ou None si le traitement a réussi. Utilisez-la pour fermer les connexions. Si open a démarré une transaction, commit uniquement lorsque error est None, et annulez-la sinon.