Tasks Snowflake : automatiser les pipelines SQL et chargement incrémental

Si tu as suivi ma série sur Snowflake jusqu'ici, on a déjà vu pas mal de briques :

• L'architecture Snowflake (storage / compute / cloud services)
• Les différents types de tables
• Les views
• Les stages
• Les file formats
• Et plus récemment les streams pour suivre ce qui a changé dans une table

On a posé les bases : ingestion, stockage, exposition des données et suivi des changements avec les streams et la question logique qui arrive derrière, c'est :

“Qui déclenche mes requêtes, et quand ?”

C'est exactement le rôle des Tasks.

C'est quoi concrètement une Task Snowflake ?

Une Task est un objet Snowflake (comme une table, une view, un stream ...) dont le job est de lancer du SQL automatiquement selon un planning ou une condition.

Tu peux y mettre :

• une requête SQL simple,
• ou l'appel d'une procédure stockée (qui elle-même peut contenir plusieurs requêtes).

Exemple de la création d'une Task

Comme toujours rien ne vaut un exemple pour comprendre le concept, on commence par créer une table de log.

CREATE OR REPLACE TABLE task_log (
  run_at   TIMESTAMP_NTZ,
  comment  STRING
);

Ensuite, on crée une Task qui ajoute une ligne dans cette table toutes les 5 minutes :

CREATE OR REPLACE TASK log_task
  WAREHOUSE = COMPUTE_WH
  SCHEDULE  = '5 MINUTE'
AS
INSERT INTO task_log
SELECT CURRENT_TIMESTAMP(), 'Task exécutée';

Pour les arguments :

• WAREHOUSE = COMPUTE_WH => on demande à la Task d'utiliser le warehouse COMPUTE_WH
• SCHEDULE = '5 MINUTE' => exécution toutes les 5 minutes,
• dans le bloc AS => le SQL à lancer à chaque run.

Nb : Au moment où on crée la Task, elle est suspendue. Il faut la démarrer :

ALTER TASK log_task RESUME;

À partir de là, Snowflake exécute l'INSERT toutes les 5 minutes.
Un SELECT * FROM task_log montre les exécutions qui s'accumulent.

Comment le compute est utilisé dans les Tasks

Pour exécuter le SQL d'une Task, Snowflake a besoin de compute.

On a principalement deux options :

1. Task “classique” avec un warehouse qu'on gère nous-mêmes
2. Task serverless, où Snowflake gère le compute.

Dans l'exemple précédent, on était en mode classique car on a indiqué WAREHOUSE = COMPUTE_WH et l'avantage c'est qu'on maîtrise précisément quel warehouse est utilisé dans la tache.

Snowflake propose aussi de laisser le moteur gérer tout ça pour nous avec des Tasks serverless. Dans ce cas, on ne donne pas de warehouse, mais juste une taille initiale, et on laisse Snowflake se débrouiller tout seul comme un grand.

Par exemple :

CREATE OR REPLACE TASK log_task_serverless
  USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
  SCHEDULE = '5 MINUTE'
AS
INSERT INTO task_log
SELECT CURRENT_TIMESTAMP(), 'Task serverless exécutée';

Ici, on laisse Snowflake s’occuper du compute pour cette Task.

Streams + Tasks : le duo pour le chargement incrémental

Dans l'article sur les streams, on a vu comment Snowflake peut nous dire :

“Voici ce qui a changé dans cette table depuis la dernière fois.”

Les Tasks sont le complément naturel :

“À quel moment on lit ces changements, et quelle est l’étape suivante du pipeline qu’on enchaîne derrière ?”

Prenons l'exemple de l'article sur les Streams :

• on a une table members qui reçoit des nouvelles lignes,
• on veut garder une autre table members_dim à jour avec ces mêmes données,

CREATE OR REPLACE TABLE members (
  id   NUMBER,
  name STRING,
  age  NUMBER
);


CREATE OR REPLACE TABLE members_dim LIKE members;


CREATE OR REPLACE STREAM members_stream
ON TABLE members
APPEND_ONLY = TRUE;

Ici, on utilise un stream APPEND_ONLY pour simplifer et ne suivre que les nouvelles lignes.

On peut alors créer une Task qui exécute régulièrement un MERGE entre le stream et members_dim :

CREATE OR REPLACE TASK members_dim_task
  WAREHOUSE = COMPUTE_WH
  SCHEDULE  = '5 MINUTE'
AS
MERGE INTO members_dim d
USING members_stream s
  ON d.id = s.id
WHEN MATCHED THEN
  UPDATE SET
    d.name = s.name,
    d.age  = s.age
WHEN NOT MATCHED THEN
  INSERT (id, name, age)
  VALUES (s.id, s.name, s.age);

On démarre la Task :

ALTER TASK members_dim_task RESUME;

Ce qui se passe ensuite est assez simple :

• des nouvelles lignes arrivent dans members
• le stream members_stream les expose comme un delta
• toutes les 5 minutes, la Task lit le stream et applique les changements sur members_dim

On obtient un chargement incrémental piloté directement depuis Snowflake.

Lancer les tasks uniquement s’il y a du delta

Pour l'optimisation, on n'a pas forcément envie que la Task s'exécute pour rien si le stream est vide.
Snowflake permet d'ajouter une condition WHEN sur la Task.

CREATE OR REPLACE TASK members_dim_task
  WAREHOUSE = COMPUTE_WH
  SCHEDULE  = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('MEMBERS_STREAM')
AS
MERGE INTO members_dim d
USING members_stream s
  ON d.id = s.id
WHEN MATCHED THEN
  UPDATE SET
    d.name = s.name,
    d.age  = s.age
WHEN NOT MATCHED THEN
  INSERT (id, name, age)
  VALUES (s.id, s.name, s.age);

La fonction SYSTEM$STREAM_HAS_DATA('MEMBERS_STREAM') renvoie TRUE s'il y a du delta à traiter.
Dans le cas contraire, la Task se “réveille”, regarde la condition, et se rendort sans lancer le MERGE ni consommer de crédits pour rien.

Chaîner plusieurs Tasks

Dans la vraie vie, c'est rare d'avoir une tâche toute seule qui vit sa vie dans son coin. La plupart du temps, on a une petite cascade de tâches qui doivent s'enchaîner.

Par exemple :

1. Une Task charge les données brutes dans des tables de staging
2. Une autre construit les modèles métiers (data marts),
3. Une troisième met à jour des agrégations ou des vues matérialisées.

Ça donne quelque chose comme :

CREATE OR REPLACE TASK task_load_raw
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 0 2 * * * Europe/Paris'
AS
CALL sp_load_raw_data();

CREATE OR REPLACE TASK task_build_marts
  WAREHOUSE = COMPUTE_WH
  AFTER task_load_raw
AS
CALL sp_build_marts();

• task_load_raw va tourner tous les jours à 2h du matin,
• task_build_marts n'a pas de SCHEDULE mais va se déclencher automatiquement après task_load_raw.

On peut chaîner plusieurs niveaux comme ça et se retrouver avec un petit DAG de Tasks.

Dans un prochain article, on pourra pousser le curseur un peu plus loin et comparer le duo Streams + Tasks avec les Dynamic Tables, pour voir dans quels cas on préfère l'un ou l'autre.

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