AccueilDocumentationInstallationMCPSDKDémoCommunautéBlog
DéveloppementIntégrationsAnalyse de données

Apache Flink

Lisez et écrivez des données CSV stockées dans le stockage objet RustFS avec Apache Flink et son plugin S3 filesystem.

Ce guide connecte Apache Flink à RustFS via le plugin S3 filesystem de Flink (flink-s3-fs-hadoop). Vous allez démarrer un cluster de session avec Docker Compose, écrire un résultat borné dans le compartiment en mode batch, puis le relire via Flink SQL. Le flux a été validé avec flink:1.20 et rustfs/rustfs-x86-musl:v2.3.1.

Vous avez besoin de Docker avec le plugin Compose. Ce déploiement est destiné aux tests d’intégration locaux, pas à la production.

Schéma d’architecture

Le plugin flink-s3-fs-hadoop enregistre le schéma s3:// pour le connecteur filesystem de Flink. Le point de terminaison, l’adressage de type chemin (path-style), HTTP non chiffré et les identifiants sont configurés via les propriétés s3.* de flink-conf.yaml (passées par FLINK_PROPERTIES).

Créer les fichiers du projet

Créez un répertoire de travail :

mkdir rustfs-flink
cd rustfs-flink

Créez un fichier d’environnement et remplacez les deux valeurs d’identifiants indicatives :

.env
RUSTFS_ACCESS_KEY=<your-access-key>
RUSTFS_SECRET_KEY=<your-secret-key>

Utilisez des identifiants dédiés pour le compartiment. N’ajoutez pas le fichier .env au dépôt (ne le versionnez pas).

Le plugin S3 est embarqué dans l’image sous /opt/flink/opt/ et doit être copié vers /opt/flink/plugins/s3fs/ pour être chargé. Préparez un répertoire local :

mkdir -p s3fs
docker create --name flink-tmp flink:1.20
docker cp flink-tmp:/opt/flink/opt/flink-s3-fs-hadoop-1.20.5.jar s3fs/
docker rm flink-tmp

Créez le fichier Compose :

compose.yaml
services:
  rustfs:
    image: rustfs/rustfs-x86-musl:v2.3.1
    environment:
      RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY}
      RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY}
      RUSTFS_VOLUMES: /data
      RUSTFS_ADDRESS: ":9000"
      RUSTFS_CONSOLE_ADDRESS: ":9001"
      RUSTFS_CONSOLE_ENABLE: "true"
    volumes:
      - rustfs-data:/data
    ports:
      - "9000:9000"
      - "9001:9001"
    healthcheck:
      test: ["CMD", "curl", "-sf", "http://127.0.0.1:9000/health"]
      interval: 10s
      timeout: 5s
      retries: 6
      start_period: 10s
    networks:
      - flink

  create-bucket:
    image: rustfs/rc:latest
    depends_on:
      rustfs:
        condition: service_healthy
    environment:
      RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY}
      RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY}
    entrypoint:
      - /bin/sh
      - -c
      - |
        until /usr/bin/rc alias set rustfs http://rustfs:9000 "$${RUSTFS_ACCESS_KEY}" "$${RUSTFS_SECRET_KEY}"; do
          echo "Waiting for RustFS..."
          sleep 2
        done
        /usr/bin/rc ls rustfs/my-bucket >/dev/null 2>&1 || /usr/bin/rc mb rustfs/my-bucket
    networks:
      - flink

  jobmanager:
    image: flink:1.20
    command: jobmanager
    environment:
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        rest.address: jobmanager
        rest.bind-address: 0.0.0.0
        s3.access-key: ${RUSTFS_ACCESS_KEY}
        s3.secret-key: ${RUSTFS_SECRET_KEY}
        s3.endpoint: http://rustfs:9000
        s3.path-style-access: true
    volumes:
      - ./s3fs:/opt/flink/plugins/s3fs:ro
    ports:
      - "8081:8081"
    depends_on:
      create-bucket:
        condition: service_completed_successfully
    networks:
      - flink

  taskmanager:
    image: flink:1.20
    command: taskmanager
    environment:
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        taskmanager.host: taskmanager
        s3.access-key: ${RUSTFS_ACCESS_KEY}
        s3.secret-key: ${RUSTFS_SECRET_KEY}
        s3.endpoint: http://rustfs:9000
        s3.path-style-access: true
    volumes:
      - ./s3fs:/opt/flink/plugins/s3fs:ro
    depends_on:
      jobmanager:
        condition: service_started
    networks:
      - flink

networks:
  flink:

volumes:
  rustfs-data:

Les propriétés s3.access-key, s3.secret-key, s3.endpoint et s3.path-style-access configurent le plugin S3 sur le JobManager et le TaskManager.

Démarrer le déploiement

Vérifiez le fichier Compose avant de démarrer les conteneurs :

docker compose config

Démarrez les services et attendez la fin de l’initialisation du compartiment :

docker compose up -d
docker compose ps -a

Écrire un résultat dans RustFS

Créez le job SQL — mode batch avec le sink filesystem :

batch.sql
SET 'execution.runtime-mode' = 'batch';

CREATE TABLE sink (
  id INT,
  payload STRING
) WITH (
  'connector' = 'filesystem',
  'path' = 's3://my-bucket/flink-out/',
  'format' = 'csv'
);

INSERT INTO sink
  VALUES (1, 'alpha'), (2, 'bravo'), (3, 'charlie'), (4, 'delta'), (5, 'echo');

Soumettez-le via le client SQL dans le JobManager :

docker compose exec jobmanager bash -c "/opt/flink/bin/sql-client.sh embedded -f /dev/stdin" < batch.sql

Le job se termine une fois toutes les lignes écrites.

Relire les données

Créez la requête de lecture — le connecteur filesystem scanne le préfixe :

read.sql
CREATE TABLE readings (
  id INT,
  payload STRING
) WITH (
  'connector' = 'filesystem',
  'path' = 's3://my-bucket/flink-out/',
  'format' = 'csv'
);

SET 'sql-client.execution.result-mode' = 'TABLEAU';

SELECT * FROM readings;
docker compose exec jobmanager bash -c "/opt/flink/bin/sql-client.sh embedded -f /dev/stdin" < read.sql
+----+-------------+--------------------------------+
| op |          id |                         payload |
+----+-------------+--------------------------------+
| +I |           1 |                           alpha |
| +I |           2 |                           bravo |
| +I |           3 |                         charlie |
| +I |           4 |                           delta |
| +I |           5 |                            echo |
+----+-------------+--------------------------------+

Vérifier les objets dans RustFS

Listez le préfixe via l’image d’initialisation du compartiment :

docker compose run --rm --entrypoint /bin/sh create-bucket -c \
  '/usr/bin/rc alias set rustfs http://rustfs:9000 "$RUSTFS_ACCESS_KEY" "$RUSTFS_SECRET_KEY" >/dev/null && /usr/bin/rc ls rustfs/my-bucket/flink-out --recursive'
[2026-09-21 01:34:52]       41 B flink-out/part-f759f9e8-3d1b-46a1-a92e-53e9b727e831-task-0-file-0

Vous pouvez également parcourir le préfixe dans la Console RustFS :

Le fichier de sortie Flink stocké dans la Console RustFS

Utiliser RustFS S3 Tables

RustFS S3 Tables fournit un catalogue REST Apache Iceberg intégré, permettant à Flink de traiter un compartiment de tables (table bucket) comme un entrepôt Iceberg géré tandis que les données restent dans RustFS. Activez un compartiment de tables et orientez le catalogue REST du connecteur Iceberg de Flink vers RustFS comme décrit dans S3 Tables : l’URI du catalogue REST est http://<rustfs-host>:9000/iceberg, le warehouse est le nom du compartiment, et les requêtes de catalogue (AWS Signature Version 4, nom de signature s3) comme l’accès aux fichiers S3 utilisent l’adressage de type chemin (path-style).

Selon la matrice de prise en charge S3 Tables, validez les versions exactes de Flink et d’Iceberg que vous déployez par rapport au catalogue avant d’adopter ce chemin en production.

Arrêter ou réinitialiser la pile

Arrêtez les conteneurs en conservant le volume de données RustFS :

docker compose down

Pour supprimer les fichiers stockés et repartir d’un volume RustFS vide, ajoutez explicitement --volumes :

docker compose down --volumes

Dépannage

No AWS Credentials provided / AccessDenied lors des écritures

Le plugin S3 lit ses identifiants depuis les propriétés s3.* de flink-conf.yaml. Vérifiez que s3.access-key, s3.secret-key, s3.endpoint et s3.path-style-access sont présents dans FLINK_PROPERTIES pour le JobManager et le TaskManager, et que le jar du plugin existe dans /opt/flink/plugins/s3fs/ sur chacun.

Le TaskManager ne résout pas le nom rustfs

Tous les conteneurs Flink et RustFS doivent partager un réseau Compose. Si RustFS est attaché à un réseau externe, connectez également les conteneurs Flink à ce réseau avant de soumettre le job.

Une écriture streaming échoue avec « Stream closed » après un échec

La reprise d’un chargement S3 (upload) en cours après un échec peut laisser le writer dans un état irrécupérable. Supprimez le préfixe de sortie du job dans le compartiment et soumettez à nouveau, ou utilisez le mode batch comme dans ce guide pour les écritures ponctuelles.

Étapes suivantes

Delta Lake

Write and read Delta tables on RustFS with delta-rs.

Hive

Store Hive table data on RustFS over S3A.

On this page

Schéma d’architecture1Créer les fichiers du projet2Démarrer le déploiement3Écrire un résultat dans RustFS4Relire les données5Vérifier les objets dans RustFS6Utiliser RustFS S3 Tables7Arrêter ou réinitialiser la pileDépannageNo AWS Credentials provided / AccessDenied lors des écrituresLe TaskManager ne résout pas le nom rustfsUne écriture streaming échoue avec « Stream closed » après un échecÉtapes suivantes