🌊 Apache Flink – Moteur de Streaming (Stateful, Event Time)

Qu'est-ce qu'Apache Flink ?

Apache Flink est un framework (cadre) et un moteur de calcul distribué open-source, conçu pour le Stateful Stream Processing (Traitement de flux "avec état").

Philosophie ("Stream-First") : Contrairement à Spark (Batch-first), Flink est un moteur de streaming "natif". Il traite les données événement par événement (ou micro-batch) avec une très faible latence.

Caractéristiques Clés (Le "Pourquoi")

• Stateful (Avec État) : (Sa force n°1) Flink est conçu pour gérer l'état (state) (ex: COUNT, SUM) localement (sur le worker), ce qui est très rapide (voir 3.1).
• Tolérance aux Pannes (Exactly-Once) : (Sa force n°2) Son mécanisme de Checkpointing (4.1) (basé sur Chandy-Lamport) garantit des sémantiques "Exactly-Once" (traitement garanti 1 fois) même en cas de panne.
• Temps (Event Time) : (Sa force n°3) Support natif de l'Event Time (Temps de l'Événement) (voir 5.1), ce qui permet de gérer les données en retard (late data) grâce aux Watermarks (5.2).
• Batch (Unifié) : Flink considère le "Batch" (traitement par lots) comme un "cas particulier" du Streaming (un flux qui se termine).

L'architecture de Flink (comme Spark/Hadoop) est Maître/Esclave. Le "Maître" (Cerveau) est le JobManager.

Rôles du JobManager

Il y a 1 JobManager par Job (Application) Flink. (Ou 1 JobManager "Cluster" en mode HA/YARN).

• Réception du Job : Reçoit le "JobGraph" (le DAG) du client (ex: flink run ...).
• Planification (Scheduling) : Transforme le "JobGraph" (logique) en "ExecutionGraph" (physique).
• Coordination (Cerveau) : Coordonne les TaskManagers (1.3) (les "Esclaves") et leur assigne des "Tasks" (tâches).
• Checkpoint Coordinator : (Rôle vital) C'est le JobManager qui initie les Checkpoints (4.1) en injectant les "Barrières" (4.2) dans le flux.
• HA (ZooKeeper) : Peut être configuré en Haute Disponibilité (HA) (Active/Standby) en utilisant ZooKeeper (similaire à HDFS HA).

Le TaskManager (ou "TaskExecutor") est l'esclave (worker). C'est le "Muscle" qui exécute le travail.

Task Slots (Unités de Parallélisme)

Un TaskManager (ex: une VM ou un Conteneur K8s) est divisé en "Task Slots" (Emplacements).

Un Slot est l'unité de ressource (CPU/RAM) de Flink. C'est (conceptuellement) 1 thread d'exécution.

Exemple : Un TaskManager (VM) avec 8 Cœurs (CPU) peut être configuré pour avoir 8 Slots.

[ TASKMANAGER (VM) ] (8 Coeurs, 32 Go RAM)
 (8 Slots)
 ├─ [Slot 1] (1 Coeur, 4Go RAM) -> (Exécute Tâche A (Job 1))
 ├─ [Slot 2] (1 Coeur, 4Go RAM) -> (Exécute Tâche B (Job 1))
 ├─ [Slot 3] (1 Coeur, 4Go RAM) -> (Exécute Tâche C (Job 2))
 └─ ...

Parallélisme

Le Parallélisme (Parallelism) de votre Job (ex: parallelism=100) définit le nombre total de Tâches (Threads) nécessaires. Flink (JobManager) distribuera ces 100 Tâches sur les Slots disponibles (ex: sur 13 TaskManagers de 8 Slots).

Data Exchange (Shuffle) : Les TaskManagers communiquent directement (via réseau/Netty) pour le "Shuffle" (ex: après un keyBy).

C'est la comparaison "classique". (Note : Spark Structured Streaming (6.1) a brouillé les pistes).

Ce sont deux moteurs de "Stateful Streaming" (3.1) très performants.

Flink (comme Spark) offre deux niveaux d'API (qui sont unifiées : on peut passer de l'une à l'autre).

(Java/Scala)
DataStream<String> stream = ...
stream
    .map(new MyMapFunction())
    .keyBy(event -> event.userID)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .reduce(new MyReduceFunction())

(Java/Scala)
Table table = tEnv.fromDataStream(stream);
Table result = table
    .window(Tumble.over(lit(5).seconds()).on(...))
    .groupBy(col("userID"), window)
    .select(col("userID"), window.end(), col("value").sum());

(SQL)
SELECT
  userID,
  TUMBLE_END(event_time, INTERVAL '5' SECOND),
  SUM(value)
FROM my_stream
GROUP BY
  TUMBLE(event_time, INTERVAL '5' SECOND),
  userID

C'est la fonctionnalité centrale de Flink.

Stateless (Sans État) : (Simple) map(), filter(). L'événement (A) est traité, puis oublié. (L'événement B ne sait rien de A).

Stateful (Avec État) : (Complexe) keyBy(), reduce(), COUNT(). Le traitement de l'événement B dépend de l'événement A.

Exemple (Compteur de Clics)

Flux d'entrée : (userA, clic), (userB, clic), (userA, clic)

Pour compter les clics (COUNT), Flink doit mémoriser (stocker) l'état (le "compte") de userA et userB.

Gestion de l'État (Local)

Flink (contrairement à d'autres) stocke cet état localement sur le TaskManager (1.3) (en RAM ou sur Disque/RocksDB (3.3)).

Avantage : L'accès à l'état (State.read(), State.write()) est extrêmement rapide (accès RAM/SSD local), au lieu d'un aller-retour réseau (ex: Redis/DB externe).

L'état (State) le plus courant est le Keyed State (État Clé). C'est un "état" qui est "scopé" (limité) à une Clé (Key) (ex: UserID).

(Il est obligatoire d'utiliser .keyBy() avant d'utiliser un Keyed State).

Types d'État (API DataStream)

(Utilisé dans un RichFlatMapFunction ou ProcessFunction)

Le State Backend (Moteur de Stockage d'État) définit où (physiquement) le TaskManager (1.3) stocke l'État (3.1) (ex: le ValueState).

Problème : L'État (3.1) est stocké localement (RAM/Disque) sur le TaskManager. Si le TaskManager (VM) crash, l'État (ex: le COUNT) est perdu.

Solution : Checkpointing (Snapshots Distants)

Le Checkpointing est le mécanisme de Flink (Tolérance aux Pannes) pour garantir Exactly-Once.

C'est un Snapshot (Instantané), coordonné (par le JobManager), de l'État (State) de tous les opérateurs (Tasks) ET de la position (Offset) dans la source (ex: Kafka).

Flux (Checkpoint)

1. (Toutes les 60 sec) Le JobManager (1.2) lance un Checkpoint (via "Barriers" (4.2)).
2. (Sur les TaskManagers) Tous les opérateurs "snapshot" (sauvegardent) leur État (State) (ex: RocksDB) vers un stockage distant durable (ex: S3, HDFS, ADLS).
3. (Source) Le connecteur Kafka "snapshot" son Offset (ex: "Partition 1, Offset 5000").
4. Les TaskManagers confirment (ACK) au JobManager.
5. Le Checkpoint (ex: #50) est "Complet".

Flux (Reprise sur Panne)

1. (Crash) TaskManager 3 meurt.
2. JobManager (HA) détecte le crash.
3. JobManager redémarre le Job (entier) depuis le dernier Checkpoint complet (#50).
4. Les nouveaux TaskManagers restaurent leur État (State) depuis S3 (RocksDB).
5. Le connecteur Kafka recommence à lire à l'Offset 5000.
6. Résultat : Aucune donnée n'est perdue (Exactly-Once).

Problème : Comment Flink prend-il un "Snapshot" (instantané) cohérent d'un système distribué (des milliers de threads) sans arrêter (pause) le flux de données ?

Solution : L'algorithme (modifié) de Chandy-Lamport, qui utilise des "Checkpoint Barriers" (Barrières).

Flux (Barrier)

1. Le JobManager (Checkpoint Coordinator) injecte une "Barrière" (ex: Barrier #50) (un message spécial) dans les Sources (ex: Kafka).
2. Cette Barrière "descend" (circule) dans le DAG (flux) avec les données (ex: entre l'événement A et B).
3. (Opérateur 1 - ex: Map) :
   • Il traite les événements (A).
   • Il reçoit Barrier #50.
   • Il "s'aligne" (align) (si plusieurs inputs).
   • Il "snapshot" (sauvegarde) son propre État (State) vers S3 (asynchrone).
   • Il transmet (forward) Barrier #50 à l'opérateur suivant.
   • Il continue de traiter les événements (B).
4. (Opérateur 2 - ex: Reduce) : (Reçoit Barrier #50, snapshot son état, transmet...).
5. (Sink - Destination) : Reçoit Barrier #50 et notifie le JobManager (ACK).

Résultat : Le flux n'est jamais stoppé ("low-overhead"). Le "Snapshot" (Checkpoint) est asynchrone et distribué.

Les deux sont des "Snapshots" de l'état (State) du Job, mais avec des objectifs différents.

Cas d'Usage (Savepoint)

Problème : Je dois mettre à jour le code (.jar) de mon Job Flink (en production) sans perdre l'État (State) (le COUNT).

Workflow (Mise à jour)

# 1. (Admin) Créer un 'Savepoint' (Snapshot) du Job (en cours)
$ flink savepoint [JobID] /path/to/savepoints/

# 2. (Admin) Arrêter l'ancien Job (v1)
$ flink cancel [JobID]

# 3. (Admin) Déployer le nouveau code (v2.jar)
$ flink run -s /path/to/savepoints/savepoint-1234 ... new-job-v2.jar

# (Flink (v2) démarre et restaure
#  automatiquement son État (State)
#  depuis le 'savepoint-1234')

Dans le streaming, "le temps" est un concept complexe. 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.
Flink (TaskManager) 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 Flink 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 un message (métadonnée) injecté dans le flux, qui dit : "Je (Flink) suis certain qu'il n'y aura plus d'événements (Event Time) avant ce timestamp."

Flux d'entrée :
(Event: 10:00:02)
(Event: 10:00:03)
(Event: 10:00:01) <-- (Désordre)
(Event: 10:00:05)
(Event: 10:00:04)

(Watermark: 10:00:00) <-- "Je suis sûr à 100%
                          qu'il n'y aura plus
                          d'événements < 10:00:00"
(Le JobManager "ferme" la fenêtre 09:55-10:00)

(Event: 10:00:07)
(Watermark: 10:00:05) <-- "Je suis sûr..."
(Le JobManager "ferme" la fenêtre 10:00-10:05)

Bounded Out-of-Orderness (Désordre Limité)

C'est la stratégie de génération de Watermark la plus courante (WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10))).

Traduction : "Générer des Watermarks basés sur le timestamp max vu (maxTs), moins 10 secondes (maxTs - 10s)". (Tolère 10 sec de retard).

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

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

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

La DataStream API est l'API "cœur" (impérative) de Flink. Une application Flink est un graphe (DAG) de DataStream.

(Java)
// 1. Obtenir l'environnement
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 2. Créer une Source (ex: Kafka)
DataStream<String> kafkaStream = env.fromSource(kafkaSource, ...);

// 3. Transformations (keyBy, window, etc.)
DataStream<WordCount> resultStream = kafkaStream
    .flatMap(...)
    .keyBy(...)
    .window(...)
    .reduce(...);

// 4. Définir un Sink (Destination)
resultStream.sinkTo(elasticsearchSink);

// 5. Exécuter le Job (Lazy)
env.execute("Mon Job Flink");

Les Connecteurs sont le "pont" (Source ou Sink) entre Flink et les systèmes externes.

• Source (Lecture) : Apache Kafka, Apache Pulsar, RabbitMQ, Kinesis, Fichiers (S3, HDFS).
• Sink (Écriture) : Apache Kafka, Elasticsearch, JDBC (SQL DBs), Cassandra, Redis, Fichiers (S3, HDFS).

C'est l'API déclarative (haut niveau) de Flink pour le streaming et le batch (similaire à Spark SQL).

-- (Exemple: Agréger un flux Kafka (clicks)
--  par fenêtre (Tumbling) de 10 secondes)

CREATE TABLE clicks (
  `user_id` BIGINT,
  `event_time` TIMESTAMP(3) METADATA FROM 'timestamp',
  WATERMARK FOR `event_time` AS `event_time` - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'clicks',
  ...
);

SELECT
  window_start,
  window_end,
  COUNT(DISTINCT user_id)
FROM
  TABLE(
    TUMBLE(TABLE clicks, DESCRIPTOR(event_time), INTERVAL '10' SECONDS)
  )
GROUP BY
  window_start, window_end;

(Java)
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

// (Convertir le DataStream en Table)
Table clicksTable = tEnv.fromDataStream(stream, $("user_id"), $("event_time").rowtime());

// (Appliquer les transformations)
Table resultTable = clicksTable
    .window(Tumble.over(lit(10).seconds()).on($("event_time")).as("w"))
    .groupBy($("w"), $("user_id"))
    .select($("user_id"), $("w").end(), $("user_id").count().as("cnt"));

FlinkCEP est une librairie (module) de Flink pour la Détection d'Événements Complexes (Complex Event Processing).

Objectif : Détecter des séquences (patterns) dans un flux d'événements.

Exemple (Détection de Fraude)

Pattern (Modèle) : "Je cherche un utilisateur (keyBy(userID)) qui fait :"

1. (Étape A) Une "Petite Transaction" (amount < 10)
2. suivie par (.next())
3. (Étape B) Une "Grosse Transaction" (amount > 1000)
4. dans les (.within()) 10 secondes.

(Java)
Pattern<Event, ?> fraudPattern = Pattern.<Event>begin("etape_A")
    .where(evt -> evt.getAmount() < 10)
    .next("etape_B")
    .where(evt -> evt.getAmount() > 1000)
    .within(Time.seconds(10));

PatternStream<Event> patternStream = CEP.pattern(inputStream.keyBy(...), fraudPattern);

DataStream<Alert> alerts = patternStream.select(
    (Map<String, List<Event>> pattern) -> {
        // (Le Pattern a matché !)
        return new Alert("Fraude Détectée !");
    }
);

Site Officiel (Apache)

• Site Web : https://flink.apache.org/
• Documentation (Stable) : https://flink.apache.org/docs/stable/
• API (JavaDoc) : https://flink.apache.org/docs/stable/api/java/

Communauté & Code

• GitHub (Code Source) : https://github.com/apache/flink
• Stack Overflow (Tag) : https://stackoverflow.com/questions/tagged/apache-flink

Fournisseurs (Flink Managé)

• Amazon Kinesis Data Analytics for Apache Flink (AWS)
• Azure HDInsight (Microsoft)
• Ververica Platform (Par les créateurs originaux de Flink)