🌊 Apache Beam – Modèle Unifié (Batch & Stream) & Runners

Qu'est-ce qu'Apache Beam ?

Apache Beam (acronyme : Batch + Stream) est un modèle de programmation (programming model) open-source, unifié, pour définir des pipelines (flux de travail) de traitement de données.

Ce n'est pas un moteur d'exécution (comme Spark ou Flink). C'est une couche d'abstraction (SDK).

Modèle Unifié (Batch & Stream)

La philosophie de Beam est que le "Batch" (lots) est juste un "cas particulier" du "Streaming" (un flux borné).

Beam fournit une seule API (le SDK Beam) pour définir des pipelines qui fonctionnent aussi bien pour des données "finies" (Batch, ex: un fichier CSV) que pour des données "infinies" (Stream, ex: un topic Kafka).

C'est le concept fondamental de Beam : la portabilité (Write-Once, Run-Anywhere).

Beam sépare la Définition (le "Quoi") de l'Exécution (le "Comment").

(Exemple PySpark : 'WordCount')
pipeline
  | 'Read' >> ReadFromText(...)
  | 'Split' >> beam.FlatMap(...)
  | 'Count' >> beam.CombinePerKey(sum)
  | 'Write' >> WriteToText(...)

(Code Beam)
    │
    ├─► (Traduction) ─► [ Spark Runner ] ─► (Exécute comme un Job Spark)
    │
    ├─► (Traduction) ─► [ Flink Runner ] ─► (Exécute comme un Job Flink)
    │
    └─► (Traduction) ─► [ Dataflow Runner ] ─► (Exécute sur GCP)

Beam n'est pas un concurrent de Spark ou Flink. C'est une abstraction au-dessus d'eux.

Le Pipeline est l'objet "racine" de toute application Beam. Il représente le graphe (DAG) complet de votre flux de travail (de la Source au Sink).

Exemple (Python SDK)

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# 1. Options (ex: le Runner à utiliser)
options = PipelineOptions([
    '--runner=DirectRunner' # (Test Local)
    # '--runner=FlinkRunner'
    # '--runner=SparkRunner'
])

# 2. Créer le Pipeline (le Graphe)
p = beam.Pipeline(options=options)

# 3. Définir le Pipeline (Source, Transforms, Sink)
(p
 | 'Read' >> beam.io.ReadFromText('input.txt')
 | 'Split' >> beam.FlatMap(...)
 | 'Write' >> beam.io.WriteToText('output.txt')
)

# 4. Exécuter le Pipeline
result = p.run()
result.wait_until_finish()

Note (Python) : L'opérateur "Pipe" | est une surcharge (syntaxe) pour la fonction .apply(). p | 'Nom' >> Transform est égal à p.apply('Nom', Transform).

Une PCollection (Parallel Collection) est l'abstraction de Beam pour représenter un ensemble de données (Dataset) distribué.

C'est l'équivalent du RDD (Spark) ou du DataStream (Flink). C'est ce qui "transite" (le "flux") entre les étapes (PTransforms).

Caractéristiques

• Immuable (Immutable) : On ne "modifie" jamais une PCollection. Une transformation (Map) crée une nouvelle PCollection.
• Distribuée : (Au runtime) La PCollection est "partitionnée" et gérée par les différents workers (Executors/TaskManagers).
• Bornée (Bounded) ou Non-Bornée (Unbounded) :
  • Bounded : (Batch) Données finies (ex: lire un fichier GCS).
  • Unbounded : (Stream) Données infinies (ex: lire un topic Kafka).

Une PTransform (Parallel Transform) est une opération (étape) dans le Pipeline.

(PCollection_In) | 'Nom_Etape' >> (PTransform) | (PCollection_Out)

Types de PTransforms

• Core Transforms (Noyau) : Les "primitives" (ex: ParDo, GroupByKey, Combine, Flatten).
• Composite Transforms (Composites) : Un "wrapper" qui regroupe plusieurs PTransforms (ex: Count() est un composite de ParDo + GroupByKey + Combine).
• I/O Transforms (Connecteurs) : Transformations "Source" (Lecture) ou "Sink" (Écriture). (ex: beam.io.ReadFromText, beam.io.WriteToKafka).

ParDo (Parallel Do) est la transformation fondamentale (le "couteau suisse") de Beam. C'est l'équivalent de flatMap (Spark/Flink).

ParDo applique une DoFn (Do Function) (votre code) à chaque élément de la PCollection.

Un DoFn peut :

• (Map) Émettre (yield) 1 élément (1-to-1).
• (Filter) Émettre 0 élément (1-to-0).
• (FlatMap) Émettre N éléments (1-to-N).

# (Note: Ce sont des raccourcis 'ParDo' composites)

# Map (1-to-1)
pcoll | 'ToUpper' >> beam.Map(lambda s: s.upper())

# Filter (1-to-0/1)
pcoll | 'Filter' >> beam.Filter(lambda s: s.startswith("Error"))

# FlatMap (1-to-N) (Ex: WordCount)
lines | 'Split' >> beam.FlatMap(lambda line: line.split(' '))

class SplitWordsFn(beam.DoFn):
    def process(self, element):
        # (element = 1 ligne d'entrée)
        # (yield = émettre N sorties)
        for word in element.split(' '):
            yield word

lines | 'Split' >> beam.ParDo(SplitWordsFn())

public class SplitWordsFn extends DoFn<String, String> {
    
    @ProcessElement
    public void processElement(@Element String line, OutputReceiver<String> out) {
        // (line = 1 ligne d'entrée)
        for (String word : line.split(" ")) {
            // (out.output = émettre N sorties)
            out.output(word);
        }
    }
}

lines.apply("Split", ParDo.of(new SplitWordsFn()));

GroupByKey (GBK) est la transformation (Wide/Shuffle) (4.3) fondamentale pour l'agrégation.

Elle prend une PCollection de Key-Value (K/V) (Paires) et la regroupe par Clé.

Flux (WordCount)

(Input PCollection)
("le", 1)
("chat", 1)
("le", 1)

   │
   │ | 'Group' >> beam.GroupByKey()
   ▼

(Output PCollection)
("le", [1, 1])
("chat", [1])

Combine (Optimisation)

Problème : GroupByKey est très coûteux (Shuffle). (Si 1 Clé a 1 Milliard de "1", l'iterable [1, 1, ...] sature la RAM du Reducer).

Solution : Utiliser Combine (ex: CombinePerKey, Count.PerKey) qui effectue une pré-agrégation (Combiner) côté "Map" (avant le Shuffle).

Une transformation Composite est une "macro" (un "wrapper") qui encapsule une chaîne de transformations (un sous-graphe) dans une seule PTransform réutilisable.

Exemple (Python) : WordCount

Le "WordCount" (Split + Count) est un PTransform composite.

# 1. Définir le PTransform Composite
#    (Hérite de beam.PTransform)
class CountWords(beam.PTransform):
    def expand(self, pcoll): # 'expand' est la méthode à implémenter
        # (pcoll = PCollection d'entrée)
        
        # (Le "sous-graphe" encapsulé)
        counts = (
            pcoll
            | 'Split' >> beam.FlatMap(lambda line: line.split(' '))
            | 'Pair' >> beam.Map(lambda word: (word, 1))
            | 'Count' >> beam.CombinePerKey(sum)
        )
        return counts

# 2. Utiliser le PTransform Composite
(p
 | 'Read' >> ReadFromText('input.txt')
 | 'Compter les Mots' >> CountWords() # (Appel de notre "macro")
 | 'Write' >> WriteToText('output.txt')
)

Le Runner est le moteur d'exécution. C'sest le "traducteur" qui prend le DAG (Graphe) logique (défini par le SDK Beam) et le traduit en un Job natif pour un moteur spécifique (Spark, Flink...).

Runners (Courants)

Le Spark Runner permet d'exécuter un pipeline Beam (Batch ou Streaming) sur un cluster Apache Spark (voir guide Spark).

Traduction (Beam -> Spark)

• Pipeline (Beam) -> Spark Job (Spark)
• PCollection (Beam) -> RDD ou DataFrame (Spark)
• ParDo (Beam) -> map / flatMap (Spark)
• GroupByKey (Beam) -> groupByKey / reduceByKey (Spark)

Exécution (CLI Python)

# (Exécuter 'mon_script_beam.py'
#  en utilisant le 'SparkRunner'
#  sur un 'master' YARN)
$ python mon_script_beam.py \
    --runner=SparkRunner \
    --spark_master=yarn \
    --deploy_mode=cluster \
    --... (options spark-submit)

Google Cloud Dataflow est le Runner "natif" et "serverless" de Beam (sur GCP). (Beam est issu des technologies internes de Google (FlumeJava, MillWheel)).

Fonctionnement (Managé)

Quand vous lancez un job avec le DataflowRunner :

1. Le SDK Beam "traduit" le pipeline en "Job Dataflow".
2. Il l'envoie à l'API Dataflow (Control Plane GCP).
3. Dataflow provisionne (auto-scale) automatiquement un pool de "Workers" (VMs) pour exécuter le job.
4. (Auto-scaling) Dataflow ajuste le nombre de workers (milliers) en fonction de la charge (Shuffle).
5. Dataflow arrête les VMs à la fin (Batch) ou continue (Streaming).

Exécution (CLI Python)

# (Exécuter sur GCP)
$ python mon_script_beam.py \
    --runner=DataflowRunner \
    --project=mon-projet-gcp \
    --region=europe-west1 \
    --temp_location=gs://mon-bucket/temp \
    --streaming (Si streaming)

Dans le streaming, "le temps" est un concept complexe. Beam (comme Flink) gère 3 notions du temps.

Exemple (Le Problème)

Un capteur IoT (iPhone) dans un ascenseur (pas de réseau) envoie un événement "Clic" (créé à 10:00:00).
L'iPhone sort de l'ascenseur (réseau) à 10:05:00.
Beam (Worker) le traite (reçoit) à 10:05:01.

• Event Time : 10:00:00 (Correct)
• Ingestion Time : 10:05:00
• Processing Time : 10:05:01

Si vous faites un window("10:00-10:01"), seul l'Event Time placera (correctement) cet événement dans la bonne fenêtre.

Problème : Si on utilise l'Event Time (5.1) et qu'on crée une fenêtre window("10:00-10:05"). Quand Beam doit-il "fermer" (calculer) cette fenêtre ? À 10:05:00 ?

Non. (Car l'événement de 10:04:00 (en retard) peut arriver à 10:05:10).

Solution : Watermarks (Filigranes)

Un Watermark (Filigrane) est la "notion du temps" (heuristique) de Beam. C'est une métadonnée qui dit : "Je (Beam) suis certain (ou 'assez certain') qu'il n'y aura plus d'événements (Event Time) avant ce timestamp."

Quand le "Watermark" (l'horloge d'Event Time) dépasse la fin d'une fenêtre (ex: 10:05:00), Beam déclenche (fires) le calcul de cette fenêtre.

Gestion du Retard (Lateness)

Les Watermarks sont (par définition) imparfaits. (L'événement de 10:04:00 peut arriver à 10:05:30, *après* la fermeture de la fenêtre 10:00-10:05).

Beam (via les Triggers (6.1)) permet de gérer ces données en retard (LATE) (ex: les ignorer (Discard), ou les inclure (Accumulate)).

Une "Fenêtre" (Window) est un "bucket" (seau) de temps (basé sur Event Time ou Processing Time) utilisé pour agréger (GroupByKey) un flux infini (Unbounded).

[-- Fenêtre 1 --] [-- Fenêtre 2 --] [-- Fenêtre 3 --]
(10:00 - 10:05)   (10:05 - 10:10)   (10:10 - 10:15)

[---- Fenêtre 1 (60s) ----]
(10:00:00 - 10:01:00)
        [---- Fenêtre 2 (60s) ----]
        (10:00:10 - 10:01:10)

Le Trigger (Déclencheur) répond à la question : "Quand (dans une fenêtre) le résultat (l'agrégation) doit-il être émis (fired) ?".

Défaut : Le trigger par défaut est AfterWatermark(). (Attend que le Watermark (5.2) passe la fin de la fenêtre).

Triggers (Exemples)

• AfterWatermark() : (Défaut) Basé sur l'Event Time (Correct).
• AfterProcessingTime() : Basé sur l'horloge (Temps réel, mais incorrect).
• AfterCount(N) : Déclencher après N éléments (ex: "tous les 100 éléments").

Combinaison & Rétraction

Les Triggers permettent des scénarios complexes : "Montre-moi un résultat (rapide) toutes les minutes (AfterProcessingTime), ET (OrFinally) le résultat correct (final) quand le Watermark arrive (AfterWatermark), ET (AfterEach) continue de mettre à jour (Accumulating) si des données en retard (LATE) arrivent."

L'API State & Timers est l'API "bas niveau" (impérative) (dans un ParDo/DoFn) pour le Stateful Streaming (3.1). (Similaire à Flink 3.2).

Elle permet à un DoFn (sur un flux "Keyed" (keyBy)) de stocker (State) et de planifier (Timer).

Exemple (Détection de "double-clic")

class DetectDoubleClickFn(beam.DoFn):
    # 1. Définir l'État (State)
    # (Stocke le timestamp du dernier clic vu)
    LAST_CLICK_STATE = beam.transforms.userstate.ValueStateSpec(
        'last_click', beam.coders.FloatCoder())

    def process(self, element,
                last_click=beam.DoFn.StateParam(LAST_CLICK_STATE)):
        
        # (Lire l'état)
        prev_click_ts = last_click.read()
        
        if prev_click_ts is not None:
            if element.timestamp - prev_click_ts < 1.0: # (1 sec)
                yield "Double Clic !" # (Émettre)
                
        # (Écrire l'état)
        last_click.write(element.timestamp)

Problème : L'API PCollection (2.2) (comme le RDD) est "non-structurée".

Solution : L'API Schema (similaire à Spark DataFrames) permet de définir une PCollection structurée (ex: PCollection ou PCollection).

Beam SQL

Une fois qu'une PCollection a un Schéma, on peut lui appliquer des transformations SQL (beam.SqlTransform) (similaire à Spark SQL).

(Python)
# (PCollection 'ventes' (avec schéma))
ventes | 'SQL Query' >> beam.SqlTransform("""
    SELECT
        userID,
        SUM(montant) AS total_ventes
    FROM
        PCOLLECTION -- (Mot-clé pour la source)
    WHERE
        montant > 10
    GROUP BY
        userID
""")

Usage : Permet aux analystes (SQL) d'écrire des pipelines Beam.

Le SDK Python (apache-beam) est le plus populaire pour la Data Science et le ML.

Installation (pip)

# Installation de base
$ pip install apache-beam

# (Optionnel) Installer les "extras"
# (Pour Google Cloud Dataflow)
$ pip install apache-beam[gcp]

# (Pour Spark Runner)
$ pip install apache-beam[spark]

# (Pour Flink Runner)
$ pip install apache-beam[flink]

# (Pour Kafka)
$ pip install apache-beam[kafka]

Usage (PySpark)

import apache_beam as beam

with beam.Pipeline() as p:
    (p | "Read" >> beam.Create([1, 2, 3])
       | "Square" >> beam.Map(lambda x: x * x)
       | "Print" >> beam.Map(print)
    )

Le SDK Java est le SDK "natif" (le plus ancien et le plus stable), souvent utilisé en production "entreprise".

Installation (Maven pom.xml)

<!-- (Dépendance de base) -->
<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-core</artifactId>
  <version>2.56.0</version>
</dependency>

<!-- (Ajouter le Runner) -->
<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-spark-3</artifactId>
  <version>2.56.0</version>
</dependency>

<!-- (ou Dataflow Runner) -->
<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
  <version>2.56.0</version>
</dependency>

Ressources officielles pour Apache Beam :