Utiliser un récepteur ForEach pour écrire chaque ligne dans les pipelines

Utilisez un récepteur ForEach pour écrire des données en streaming vers des destinations qui n’ont pas de récepteur intégré, comme une API REST externe ou un magasin clé-valeur.

Le sink ForEach traite un stream ligne par ligne. Pour chaque ligne, le pipeline appelle une fonction Python personnalisée ou un objet d'écriture, semblable au sink foreach dans Apache Spark Structured Streaming.

remarque

Le sink ForEach décrit sur cette page est destiné aux pipelines. Apache Spark Structured Streaming prend aussi en charge foreach. Pour plus d’informations sur l’écriture de sorties personnalisées dans Structured Streaming, voir Utiliser foreachBatch pour écrire dans des sinks de données arbitraires.

Quand utiliser une destination ForEach

Utilisez un sink ForEach quand vous devez écrire chaque enregistrement individuellement vers une destination qui n’a pas de format de sink intégré, comme l’appel d’une API externe ou la mise à jour d’un magasin clé-valeur pour chaque enregistrement. Utilisez plutôt un sink ForEachBatch quand votre logique fonctionne sur un micro-batch complet, comme la fusion dans une table Delta Lake ou l’écriture dans plusieurs destinations avec des opérations DataFrame.

Ce tableau compare les deux récepteurs :

Caractéristique	ForEach puits	ForEachBatch puits
Unité de traitement	Une ligne à la fois	Un DataFrame de micro-batch à la fois
Où votre code s'exécute	Sur les exécuteurs	Sur le driver
API Python	@dp.foreach_sink()	@dp.foreach_batch_sink()

Caractéristique	ForEach puits	ForEachBatch puits
Unité de traitement	Une ligne à la fois	Un DataFrame de micro-batch à la fois
Où votre code s'exécute	Sur les exécuteurs	Sur le driver
API Python	@dp.foreach_sink()	@dp.foreach_batch_sink()

Pour plus d’informations sur les récepteurs intégrés, consultez Récepteurs dans les LakeFlow Pipelines. Pour la référence de l’API Python @dp.foreach_sink(), consultez foreach_sink.

Définir un puits ForEach

Vous pouvez définir une destination ForEach à partir d’une fonction ou d’un objet writer.

Pour définir la destination à partir d’une fonction, appliquez @dp.foreach_sink() à la fonction. Le pipeline appelle la fonction pour chaque ligne. Si vous omettez name, la destination utilise le nom de la fonction :

Python

from pyspark import pipelines as dp

@dp.foreach_sink()
def my_row_sink(row):
    # Custom per-row logic here.
    ...

Pour définir le récepteur à partir d’un objet d’écriture, passez l’objet avec l’argument de mot clé writer. L’écriture doit implémenter process(row) et peut implémenter open(partition_id, epoch_id) et close(error). Utilisez l’écriture lorsque vous souhaitez ouvrir une connexion pour chaque partition et la réutiliser pour chaque ligne. Si vous omettez name, le récepteur utilise le nom de classe de l’écriture, en minuscules :

Python

from pyspark import pipelines as dp

class MyRowWriter:
    def open(self, partition_id, epoch_id):
        # Open a connection for this partition.
        return True

    def process(self, row):
        # Write the row by using the connection.
        ...

    def close(self, error):
        # Close the connection.
        ...

dp.foreach_sink(name="my_row_sink", writer=MyRowWriter())

Pour obtenir des détails sur chaque méthode, consultez Méthodes d'écriture.

Entrées des Logs des événements

Lorsque vous définissez un récepteur ForEach, le log d’événements du pipeline enregistre un événement SinkDefinition avec "format": "foreach" pour le récepteur.

Écrire vers un récepteur ForEach

Vous pouvez écrire vers un sink ForEach depuis un flux d'ajout ou un flux de mise à jour.

Écrire avec un flux d’ajout

Pour écrire dans un sink ForEach à partir d'un flux append, référencez le nom du sink comme target:

Python

@dp.append_flow(target="my_row_sink")
def my_flow():
    return spark.readStream.table("my_catalog.my_schema.source_table")

Vous pouvez définir plusieurs flux d'ajout qui ciblent le même récepteur ForEach. Chaque flux conserve son propre point de contrôle.

Écrire avec un flux de mise à jour

Pour écrire vers un récepteur ForEach à partir d’un flux de mise à jour, utilisez le décorateur @dp.update_flow :

Python

@dp.update_flow(target="my_row_sink")
def my_stateful_flow():
    return spark.readStream.table("my_catalog.my_schema.events").groupBy("user_id").count()

Un flux de mise à jour peut renvoyer la même clé au récepteur dans un micro-batch ultérieur avec une nouvelle valeur. Pour remplacer l'ancienne valeur, effectuez un upsert de chaque ligne dans votre destination par clé au lieu de l'insérer. Dans l'exemple précédent, utilisez user_id comme clé, de sorte que chaque nouveau compte pour un utilisateur remplace le compte précédent.

Pour le traitement à faible latence, un flux de mise à jour qui écrit dans un récepteur ForEach peut s’exécuter en mode temps réel. Consultez Utiliser le mode temps réel dans les LakeFlow Pipelines.

Importer des bibliothèques et des dépendances sur les exécuteurs

Votre fonction ou votre rédacteur s’exécute sur les executors, pas sur le driver. Suivez les pratiques suivantes :

• Ajoutez les packages dont votre code a besoin à l’environnement de pipeline. Les packages de l’environnement de pipeline sont disponibles pour votre code sur les exécuteurs. Consultez Gérer les dépendances Python pour les pipelines.
• Importez les bibliothèques que votre code utilise à l’intérieur de la fonction ou des méthodes d’écriture, pas seulement en haut du fichier source du pipeline.
• Ne vous appuyez pas sur des objets que vous créez sur le driver, par exemple un client ou une connexion que vous ouvrez au début du fichier. Créez-les à l'intérieur de la fonction, ou dans la méthode open du writer.

Traiter les échecs et les lignes en double

Si votre code génère une exception, le flux échoue et le journal des événements du pipeline enregistre un événement d'erreur qui inclut le message d'exception.

Lorsqu’un micro-batch est relancé après un échec, le sink peut recevoir certaines lignes à nouveau. Rendez vos écritures par ligne idempotentes, par exemple en utilisant une clé unique de la ligne pour effectuer une upsert dans la destination. Voir Garanties de traitement dans les LakeFlow Pipelines.

Si vous utilisez un objet writer, open reçoit un partition_id et un epoch_id. Ensemble, ces IDs identifient les lignes d'une partition dans un micro-batch, afin que vous puissiez les utiliser pour dédoublonner les lignes. Une fonction ne reçoit pas ces IDs.

Les ID ne sont uniques que dans le point de contrôle d'un flux. La même paire peut se répéter dans les cas suivants :

• Plusieurs flux écrivent dans le même récepteur.
• Un refresh complet Reset les points de contrôle.

Pour dédoublonner avec ces identifiants, écrivez vers le récepteur à partir d’un flux unique et effacez votre état de dédoublonnement lorsque vous exécutez un refresh complet.

Actualisation complète refresh

Lors d'un refresh complet, le pipeline reset le checkpoint de chaque flux qui écrit dans le récepteur ForEach. Le pipeline ne suit ni ne nettoie les données écrites par le récepteur, vous devez donc gérer la rétention et le nettoyage pour votre destination. Si vous avez besoin d'un clean start, supprimez vous-même ces données.

Limitations

• Métrique de ligne : le pipeline ne signale pas le nombre de lignes qu’un récepteur ForEach écrit.
• Appels par ligne : si votre code appelle un système externe pour chaque ligne, le système externe peut limiter le throughput. Pour réduire la surcharge, ouvrez des connexions dans la méthode open de l’enregistreur et réutilisez-les pour chaque ligne, ou utilisez un récepteur ForEachBatch pour écrire en masse.

Exemples

Ces exemples montrent des méthodes courantes pour utiliser un récepteur ForEach.

Exemple de syntaxe de base

Cet exemple définit un récepteur ForEach à partir d’une fonction et y écrit à partir d’un flux d’ajout :

Python

from pyspark import pipelines as dp

# Create a ForEach sink
@dp.foreach_sink(name="my_foreach_sink")
def handle_row(row):
    # Custom per-row logic here.
    ...

# Create source data for the example
@dp.table()
def example_source_data():
    return spark.range(5)

# Write to the sink from an append flow
@dp.append_flow(target="my_foreach_sink")
def my_flow():
    return spark.readStream.table("example_source_data")

Écrire chaque ligne dans une API externe

Cet exemple utilise un objet writer pour envoyer chaque commande à une API REST externe. L’objet writer ouvre une session HTTP pour chaque partition. Il utilise l’ID de la commande dans l’URL, de sorte qu’une ligne retentée met à jour le même enregistrement au lieu de créer un doublon :

Python

from pyspark import pipelines as dp

class OrdersApiWriter:
    def open(self, partition_id, epoch_id):
        import requests  # Import on the executor

        self.session = requests.Session()
        return True

    def process(self, row):
        response = self.session.put(
            f"https://api.example.com/orders/{row.order_id}",
            json={"order_id": row.order_id, "status": row.status},
            timeout=10,
        )
        # Raise on HTTP errors so that the flow fails instead of dropping the row.
        response.raise_for_status()

    def close(self, error):
        self.session.close()

dp.foreach_sink(name="orders_api_sink", writer=OrdersApiWriter())

@dp.append_flow(target="orders_api_sink")
def orders_to_api():
    return spark.readStream.table("my_catalog.my_schema.orders")

Écrire vers un récepteur à partir de plusieurs flux

Cet exemple écrit deux sources dans le même récepteur ForEach :

Python

from pyspark import pipelines as dp

@dp.foreach_sink(name="events_sink")
def handle_event(row):
    # Custom per-row logic here.
    ...

@dp.append_flow(target="events_sink")
def web_events():
    return spark.readStream.table("my_catalog.my_schema.web_events")

@dp.append_flow(target="events_sink")
def mobile_events():
    return spark.readStream.table("my_catalog.my_schema.mobile_events")

Ressources supplémentaires

• foreach_sink
• Utilisez ForEachBatch pour écrire vers des data sinks arbitraires dans les pipelines
• Utilisez foreachBatch pour écrire vers des récepteurs de données arbitraires
• Garanties de traitement dans les LakeFlow Pipelines
• Utiliser le mode temps réel dans les LakeFlow Pipelines