Snowpipe : ingestion continue des données dans Snowflake

Dans les articles précédents, on a chargé les données avec un bon vieux :

COPY INTO sales_raw
FROM @stg_sales
FILE_FORMAT = (FORMAT_NAME = 'CSV_SALES_FORMAT');

Ça marche très bien pour du batch quand on lance la commande car tout simplement on maîtrise quand on veut charger les données.

Mais dès qu'on a des fichiers qui tombent toute la journée dans un bucket, ou qu'on veut quelque chose de plus proche du temps réel, relancer un COPY à la main (ou même via une Task) n'est pas la meilleure option.

C'est exactement là que Snowflake sort Snowpipe et les Pipes.

C'est quoi un Pipe / Snowpipe concrètement ?

Concrètement, un pipe Snowflake, c'est juste un objet qui encapsule une commande COPY INTO avec toutes ses options (source @stage, file format, etc...).

Snowpipe, c'est le service managé de Snowflake qui surveille un stage (ou des events cloud) et déclenche automatiquement ce pipe dès qu'il détecte de nouveaux fichiers.

Sans Pipe, on est dans le mode :

"J'écris un COPY et je le lance moi-même ou avec une Task."

Avec un Pipe / Snowpipe, on passe au mode :

"J'écris une fois la logique de chargement et Snowflake s'occupe de déclencher le COPY quand de nouveaux fichiers arrivent."

Tout ça repose sur ce qu'on a déjà vu :

• Le stage => où sont les fichiers,
• Le file format => comment les lire,
• Le Pipe => quoi charger et vers quelle table (la requête COPY INTO préconfigurée).

Un exemple simple pour voir comment ça marche

On part d'un cas classique :

CREATE OR REPLACE TABLE sales_raw (
  id        NUMBER,
  sale_date DATE,
  amount    NUMBER(10,2),
  country   STRING
);

CREATE OR REPLACE STAGE stg_sales
  FILE_FORMAT = (FORMAT_NAME = 'CSV_SALES_FORMAT');

Sans Snowpipe, on ferait régulièrement :

COPY INTO sales_raw
FROM @stg_sales;

Avec Snowpipe, on crée un Pipe qui encapsule ce COPY :

CREATE OR REPLACE PIPE sales_pipe
AS
COPY INTO sales_raw
FROM @stg_sales;

Là, on vient de dire à Snowflake : "Dès que tu vois des nouveaux fichiers dans @stg_sales, tu les charges dans sales_raw avec ces règles-là."

Sur un stage interne, on peut déjà orchestrer l'ingestion en demandant à Snowpipe de scanner les nouveaux fichiers via :

ALTER PIPE sales_pipe REFRESH;

Ça dit à Snowflake => "Regarde les nouveaux fichiers dans ce stage et ingère-les via ce Pipe".

AUTO_INGEST : le mode automatique

Le vrai intérêt de Snowpipe, c'est qu'il peut se déclencher tout seul sans qu'on ait besoin de lancer un REFRESH manuellement. C'est le paramètre AUTO_INGEST = TRUE qui active ce mode.

CREATE OR REPLACE PIPE sales_pipe
  AUTO_INGEST = TRUE
AS
COPY INTO sales_raw
FROM @stg_s3_raw/sales/
FILE_FORMAT = (FORMAT_NAME = 'CSV_SALES_FORMAT');

Le principe est simple et c'est dès qu'un fichier arrive dans le stage, une notification cloud prévient Snowpipe et il déclenche le chargement automatiquement.

En pratique, la notification dépend du cloud provider :

• AWS : tu configures un event notification S3 qui envoie un message vers un topic SNS. Snowpipe écoute ce topic.
• Azure : tu passes par Event Grid. Chaque blob créé dans le container déclenche une notification vers Snowpipe.
• GCP : c'est Pub/Sub qui fait le relais entre Cloud Storage et Snowpipe.

Pour AWS par exemple, une fois le pipe créé avec AUTO_INGEST = TRUE, tu récupères l'ARN de la queue SQS que Snowflake a créée automatiquement :

SHOW PIPES;
-- La colonne notification_channel te donne l'ARN SQS

Ensuite tu configures l'event notification côté S3 pour envoyer les événements s3:ObjectCreated:* vers cette queue SQS. À partir de là, c'est complètement automatique : un fichier tombe dans S3 → S3 notifie SQS → Snowpipe détecte le fichier → le COPY se lance.

Sans AUTO_INGEST , tu dois déclencher le chargement toi-même avec ALTER PIPE ... REFRESH ou via l'API REST Snowpipe (insertFiles).

Compute serverless : comment Snowpipe est facturé

C'est un point important à comprendre. Snowpipe n'utilise pas ton warehouse. Il utilise du compute serverless géré par Snowflake. Ça veut dire que tu n'as pas besoin de dimensionner un warehouse, de gérer l'auto-suspend ou de t'inquiéter de la taille du cluster.

Snowflake alloue des ressources de calcul à la demande et te facture en crédits Snowpipe à la seconde. Le coût dépend du volume de fichiers et de données traités, pas d'un warehouse qui tourne en continu.

Concrètement, ça change la logique de coûts par rapport à un COPY INTO classique :

Le piège c'est que si tu as un gros batch de 500 fichiers qui tombent d'un coup une fois par jour, Snowpipe sera probablement plus cher qu'un COPY INTO dans une Task qui allume un warehouse Medium pendant 2 minutes. Par contre, si tu as des fichiers qui arrivent au fil de l'eau toute la journée, Snowpipe est clairement plus adapté et souvent moins cher que de maintenir un warehouse allumé en permanence.

Gestion des erreurs : ON_ERROR dans Snowpipe

C'est une différence subtile mais importante. Quand tu fais un COPY INTO classique, le comportement par défaut en cas d'erreur est ABORT_STATEMENT et donc dès qu'une ligne pose problème, tout le chargement s'arrête.

Avec Snowpipe, le comportement par défaut est SKIP_FILE , donc si un fichier contient des erreurs, Snowpipe le saute et passe au suivant.

C'est logique car Snowpipe tourne en continu, il ne peut pas se permettre de tout bloquer à cause d'un fichier mal formaté.

Tu peux évidemment personnaliser ce comportement dans la définition du pipe :

CREATE OR REPLACE PIPE sales_pipe
  AUTO_INGEST = TRUE
AS
COPY INTO sales_raw
FROM @stg_s3_raw/sales/
FILE_FORMAT = (FORMAT_NAME = 'CSV_SALES_FORMAT')
ON_ERROR = 'CONTINUE';  -- skip les lignes en erreur, pas le fichier entier

Les options sont les mêmes que pour COPY INTO :

CONTINUE (skip les lignes en erreur), SKIP_FILE (skip le fichier entier, c'est la valeur par défaut dans Snowpipe), SKIP_FILE_n (skip si plus de n erreurs), SKIP_FILE_n% (skip si le pourcentage d'erreurs dépasse n%).

Surveiller Snowpipe : SYSTEM$PIPE_STATUS et COPY_HISTORY

Une fois que ton pipe tourne en production, tu veux savoir ce qui se passe. Deux outils essentiels.

SYSTEM$PIPE_STATUS

Cette fonction te donne l'état en temps réel du pipe et donc savoir combien de fichiers sont en attente de chargement, combien ont été traités, si le pipe est en pause ou actif.

SELECT SYSTEM$PIPE_STATUS('sales_pipe');

Le résultat est un JSON qui contient entre autres pendingFileCount (fichiers en attente), executionState (RUNNING, PAUSED, STALLED...) et lastIngestedTimestamp (dernier chargement réussi). C'est le premier endroit où regarder si tu suspectes un problème d'ingestion.

COPY_HISTORY

Pour l'historique détaillé de chaque chargement, tu utilises la vue COPY_HISTORY. Tu y trouves le nom du fichier, le nombre de lignes chargées, les erreurs, le statut, la durée, etc...

-- Historique des chargements Snowpipe des dernières 24h
SELECT
  pipe_name,
  file_name,
  row_count,
  row_parsed,
  error_count,
  first_error_message,
  pipe_received_time,
  last_load_time
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'sales_raw',
  START_TIME => DATEADD(hour, -24, CURRENT_TIMESTAMP())
))
ORDER BY pipe_received_time DESC;

Si tu veux l'historique sur une plus longue période (jusqu'à 365 jours), tu passes par SNOWFLAKE.ACCOUNT_USAGE.COPY_HISTORY au lieu de INFORMATION_SCHEMA, mais avec une latence de 1 à 2 heures sur les données (comme pour toutes les vues ACCOUNT_USAGE, voir l'article FinOps Snowflake).

Gérer un pipe : pause, resume, drop

-- Mettre en pause (le pipe arrête de traiter les fichiers)
ALTER PIPE sales_pipe SET PIPE_EXECUTION_PAUSED = TRUE;

-- Reprendre
ALTER PIPE sales_pipe SET PIPE_EXECUTION_PAUSED = FALSE;

-- Supprimer le pipe (les données déjà chargées restent en table)
DROP PIPE sales_pipe;

Point important quand tu mets un pipe en pause puis que tu le reprends, Snowpipe traite les fichiers qui se sont accumulés pendant la pause et donc il ne les perd pas.

Le mécanisme anti-doublon : l'historique des 64 jours

Comme pour COPY INTO, Snowpipe garde un historique des fichiers déjà chargés pendant 64 jours. Si un fichier a déjà été ingéré, Snowpipe ne le rechargera pas, même si tu fais un REFRESH.

C'est un mécanisme de sécurité contre les doublons, et il fonctionne sur le nom et le hash du fichier. Si tu veux forcer le rechargement (par exemple après une correction du fichier), il faut renommer le fichier ou recréer le pipe.

Attention, ce n'est pas configurable. 64 jours, c'est fixe. Après ce délai, si le même fichier est toujours dans le stage, Snowpipe pourrait le recharger.

Snowpipe vs Task + COPY : quand utiliser quoi ?

Pour faire simple :

• Si tu as des fichiers qui arrivent en continu ou de manière imprévisible → Snowpipe est le bon choix. Tu configures une fois et tu n'y touches plus.
• Si tu es sur un batch bien cadré, à une heure précise, avec d'autres étapes derrière (streams, transformations etc...) → un COPY dans une Task reste la meilleure option.

Les deux approches ne s'excluent pas du tout. Tu peux très bien avoir Snowpipe pour alimenter une table brute en continu, puis derrière des streams + tasks ou des Dynamic Tables pour alimenter le reste du pipeline.

Snowpipe Streaming

Snowpipe Streaming est une variante de Snowpipe pensée pour les cas où même la latence de quelques minutes du Snowpipe classique n'est pas suffisante. Au lieu de charger depuis des fichiers sur un stage, Snowpipe Streaming permet d'insérer des lignes directement via une API (SDK Java), sans passer par un fichier intermédiaire.

C'est adapté aux cas d'usage comme l'ingestion de logs applicatifs, d'événements IoT ou un flux Kafka en quasi temps réel (latence de quelques secondes). Le pricing est différent aussi : tu paies au volume de données ingérées, pas en crédits Snowpipe classiques.

Pour l'instant, le SDK est disponible uniquement en Java. Concrètement, Snowpipe Streaming est utilisé en production surtout via le connecteur Kafka pour Snowflake.

Où se place Snowpipe dans la formation ?

Snowpipe, c'est une brique de plus dans le puzzle de la formation Snowflake que je construis sur le blog.

Si tu découvres la série, tu peux reprendre le fil depuis le début :

• Architecture Snowflake : storage, compute et Cloud Services pour comprendre comment Snowflake est construit (storage, compute, cloud services)
• Types de tables Snowflake : permanent, transient, temporary et external pour choisir le bon type de table en fonction de la durée de vie et du coût de tes données.
• Types de vues Snowflake : standard, secure et materialized pour exposer tes données proprement, sécuriser ce qu'il faut et optimiser les cas de lecture lourde.
• Stages Snowflake : table stage, user stage et named stage pour savoir où déposer tes fichiers et comment les organiser avant de les charger.
• Snowflake file format : CSV, JSON, Parquet... pour apprendre à dire à Snowflake comment lire tes fichiers.
• Streams Snowflake : suivre les changements de données (CDC) pour capter ce qui change dans tes tables et alimenter le reste de ton pipeline.
• Tasks Snowflake : automatiser les pipelines SQL et le chargement incrémental pour planifier l'exécution de ton SQL, orchestrer tes chargements.
• Dynamic Table Snowflake : créer des pipelines de chargement incrémental

Avec tout ça, Snowpipe vient compléter la chaîne. Tu déposes des fichiers sur un stage → Snowpipe les charge en continu → streams + tasks ou Dynamic Tables prennent le relais pour tenir ton modèle à jour.

Aller plus loin : Formation Snowflake

J'ai regroupé tous mes articles Snowflake dans un parcours complet.

👉 Accéder à la Formation Snowflake

Vous voulez que je vous accompagne sur votre projet data (Snowflake, ingestion, modélisation, performance, coûts, gouvernance) ?

👉 Réserver un appel de 30 minutes