Étape IV : Exemple de pipeline complet

On assemble les quatre étapes précédentes dans un scénario réaliste de réponse à la demande :

« On nous demande la valeur moyenne d’un indicateur par région, pour l’année 2024. La donnée source vit dans PostgreSQL. Le livrable doit être un fichier Parquet déposé sur S3. »

Le script ci-dessous reprend, dans l’ordre :

connexion à DuckDB ;
connexion à PostgreSQL et export en Parquet ;
manipulation des données (ici en dplyr, mais le SQL natif fonctionnerait aussi bien) ;
export du résultat final sur S3 ;
fermeture de la connexion.

library(DBI)
library(duckdb)
library(dplyr)
library(glue)

# --- Étape 0 : connexion -----------------------------------------------

con <- DBI::dbConnect(duckdb::duckdb())

# --- Configuration S3 (nécessaire pour les étapes I et III) ------------

DBI::dbExecute(con, "INSTALL httpfs; LOAD httpfs;")

DBI::dbExecute(con, glue::glue("
  CREATE OR REPLACE SECRET secret_s3 (
    TYPE s3,
    PROVIDER credential_chain,
    ENDPOINT '{Sys.getenv(\"AWS_S3_ENDPOINT\")}',
    REGION '{Sys.getenv(\"AWS_DEFAULT_REGION\")}',
    URL_STYLE 'path'
  );
"))

# --- Étape I : chargement (PostgreSQL -> Parquet) -----------------------

DBI::dbExecute(con, "INSTALL postgres; LOAD postgres;")

# Le mot de passe est lu depuis une variable d'environnement car ce script
# est pensé pour tourner sans interaction (en session interactive, on peut
# le remplacer par rstudioapi::askForPassword())
DBI::dbExecute(con, glue::glue("
  CREATE OR REPLACE SECRET secret_pg (
    TYPE postgres,
    HOST '{Sys.getenv(\"PG_HOST\")}',
    PORT {Sys.getenv(\"PG_PORT\")},
    DATABASE '{Sys.getenv(\"PG_DB\")}',
    USER '{Sys.getenv(\"PG_USER\")}',
    PASSWORD '{Sys.getenv(\"PG_PASSWORD\")}'
  );
  ATTACH '' AS pg (TYPE postgres, SECRET secret_pg, READ_ONLY);
"))

# On extrait la table source vers un Parquet de travail sur S3
DBI::dbExecute(con, "
  COPY pg.ma_table
  TO 's3://mon-bucket/mon-etude/brut/ma_table.parquet'
  (FORMAT parquet)
")

# On la relit pour la suite du pipeline (vue paresseuse, rien n'est chargé en mémoire)
DBI::dbExecute(con, "
  CREATE OR REPLACE VIEW ma_table AS
  SELECT * FROM 's3://mon-bucket/mon-etude/brut/ma_table.parquet'
")

# PostgreSQL n'est plus nécessaire pour la suite : on ferme la connexion
DBI::dbExecute(con, "DETACH pg")

# --- Étape II : manipulation (ici en dplyr) -----------------------------

resultat <- dplyr::tbl(con, "ma_table") |>
  dplyr::filter(annee == 2024) |>
  dplyr::group_by(region) |>
  dplyr::summarise(valeur_moyenne = mean(valeur, na.rm = TRUE)) |>
  dplyr::collect()

# On enregistre le résultat comme vue DuckDB pour pouvoir l'exporter en SQL
# (duckdb_register ne copie pas les données, contrairement à dbWriteTable)
duckdb::duckdb_register(con, "resultat_final", resultat)

# --- Étape III : export du livrable -------------------------------------

DBI::dbExecute(con, "
  COPY resultat_final
  TO 's3://mon-bucket/mon-etude/livrable/resultat_regions_2024.parquet'
  (FORMAT parquet)
")

duckdb::duckdb_unregister(con, "resultat_final")

# --- Fermeture -----------------------------------------------------------

DBI::dbDisconnect(con, shutdown = TRUE)

Note

Ce script mélange volontairement dplyr (étape II) et SQL natif (étapes I et III) pour montrer que les deux cohabitent sans problème dans un même pipeline : c’est toujours DuckDB qui exécute le calcul.

Astuce

Si toute la manipulation avait été écrite en SQL natif, l’étape II et l’enregistrement intermédiaire via duckdb::duckdb_register() auraient pu être remplacés par une seule requête COPY (SELECT ...) TO ..., comme montré à l’étape III.

Pour aller plus loin

Documentation DuckDB — la référence complète.
Client R — toutes les fonctions DBI et dbplyr disponibles.
r.duckdb.org — manuel de référence du package R.

Retour au sommet