Accueil›Blog›Test technique Snowflake avance : Streams, Tasks, Dynamic Tables

Guide recrutement data

Test technique Snowflake avance : Streams, Tasks, Dynamic Tables

Snowflake n est pas juste un data warehouse SQL. En entretien Senior, on eval...

Data Builder·Juin 2025·7 min·Data Engineer · Analytics Engineer

Sommaire

Streams et CDC
Tasks et orchestration
Dynamic Tables
Zero-Copy Cloning
Snowpark Python
Optimisation des couts
Grille

1Streams : CDC natif dans Snowflake

Question discriminante

Qu'est-ce qu'un Stream Snowflake ? Quelle est la différence entre APPEND_ONLY et DEFAULT ?

-- Créer un stream sur une table source
CREATE STREAM orders_stream ON TABLE orders
    APPEND_ONLY = FALSE;  -- capture INSERT + UPDATE + DELETE

-- APPEND_ONLY = TRUE : capture uniquement les INSERT (plus performant)
-- APPEND_ONLY = FALSE : capture INSERT + UPDATE + DELETE via METADATA$ACTION

-- Consommer le stream dans un MERGE
MERGE INTO orders_target AS t
USING (
    SELECT order_id, customer_id, amount,
           METADATA$ACTION,      -- INSERT ou DELETE
           METADATA$ISUPDATE     -- True si c'est un UPDATE (représenté par DELETE+INSERT)
    FROM orders_stream
) AS s ON t.order_id = s.order_id
WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' AND NOT s.METADATA$ISUPDATE 
    THEN DELETE
WHEN MATCHED AND s.METADATA$ACTION = 'INSERT' AND s.METADATA$ISUPDATE 
    THEN UPDATE SET t.amount = s.amount, t.customer_id = s.customer_id
WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT' 
    THEN INSERT (order_id, customer_id, amount) VALUES (s.order_id, s.customer_id, s.amount);

-- Vérifier si le stream a des données à consommer
SELECT SYSTEM$STREAM_HAS_DATA('orders_stream');

Streams sont des pointeurs — ils pointent sur un offset dans l'historique de la table (comme Kafka). Consommer le stream avance l'offset
Multiple streams — plusieurs streams peuvent pointer sur la même table indépendamment
Expiration — si un stream n'est pas consommé dans la fenêtre Time Travel (7-90 jours), il expire et doit être recréé

2Tasks : orchestration native Snowflake

Question discriminante

Comment combinez-vous Streams et Tasks pour un pipeline CDC automatique ?

-- Task planifiée toutes les heures
CREATE OR REPLACE TASK refresh_aggregates
    WAREHOUSE = 'ETL_WH'
    SCHEDULE = 'USING CRON 0 * * * * UTC'
    ERROR_INTEGRATION = 'my_notification_int'  -- alerte en cas d'erreur
AS
    CALL refresh_daily_revenue();

ALTER TASK refresh_aggregates RESUME;

-- Task déclenchée automatiquement par un Stream (quand nouvelles données)
CREATE OR REPLACE TASK process_new_orders
    WAREHOUSE = 'ETL_WH'
    AFTER refresh_aggregates  -- task parente (DAG de tasks)
    WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
    MERGE INTO orders_processed AS t
    USING orders_stream AS s ON t.order_id = s.order_id
    WHEN MATCHED THEN UPDATE SET t.amount = s.amount
    WHEN NOT MATCHED THEN INSERT VALUES (s.order_id, s.customer_id, s.amount);

-- Surveiller les tasks
SELECT *
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START => DATEADD('hour', -24, CURRENT_TIMESTAMP)
))
ORDER BY SCHEDULED_TIME DESC;

3Dynamic Tables : incrémental automatique

Question discriminante

Qu'est-ce qu'une Dynamic Table ? Quand l'utiliser à la place d'un modèle dbt ?

-- Créer une Dynamic Table
CREATE OR REPLACE DYNAMIC TABLE analytics.daily_revenue
    TARGET_LAG = '1 hour'  -- fraîcheur souhaitée
    WAREHOUSE = ETL_WH
AS
SELECT
    DATE(order_date) AS day,
    region,
    SUM(amount) AS revenue,
    COUNT(*) AS order_count
FROM raw.orders
GROUP BY 1, 2;

-- Snowflake gère automatiquement le refresh incrémental
-- Pas besoin de scheduler externe ni de logique incremental manuelle

-- Vérifier le statut
SHOW DYNAMIC TABLES LIKE 'daily_revenue';

TARGET_LAG — délai acceptable entre les sources et la Dynamic Table. '1 hour' = données fraîches de moins d'1h
vs Materialized Views — les Dynamic Tables supportent les JOINs complexes, les CTEs, les sous-requêtes. Bien plus expressif
vs dbt incremental — les Dynamic Tables sont automatiques (pas de scheduler). dbt incremental nécessite une orchestration externe
Coût — Snowflake facture le compute de refresh. Un TARGET_LAG court = plus de refreshes = plus de coût

4Zero-Copy Cloning

Question discriminante

Qu'est-ce que le Zero-Copy Cloning ? Comment l'utilisez-vous en pratique ?

-- Cloner une base entière en quelques secondes, sans copier les données
CREATE DATABASE staging_clone CLONE production;
-- Les données ne sont PAS dupliquées. Partage des micro-partitions.
-- Écriture en staging → copie uniquement des partitions modifiées (copy-on-write)

-- Use cases :
-- 1. Test de migration sans risque
-- 2. Environnement de dev avec données de production
-- 3. Snapshot rapide avant une opération risquée

-- Time Travel : requêter les données passées
SELECT * FROM orders AT(TIMESTAMP => '2025-01-01 00:00:00'::TIMESTAMP_TZ);
SELECT * FROM orders BEFORE(STATEMENT => 'last_bad_query_id');

-- Restaurer une table depuis Time Travel
CREATE TABLE orders_restored CLONE orders
    AT(TIMESTAMP => '2025-01-01 00:00:00'::TIMESTAMP_TZ);

-- Undrop : restaurer une table supprimée (dans la fenêtre Time Travel)
DROP TABLE orders;
UNDROP TABLE orders;  -- fonctionne pendant 7 à 90 jours selon la config

5Snowpark Python : Python dans Snowflake

Question discriminante

Qu'est-ce que Snowpark ? Quand l'utiliser à la place d'un script Python externe ?

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, when

session = Session.builder.configs({
    "account": "mon-compte",
    "user": "data_engineer",
    "private_key": load_private_key()
}).create()

# DataFrame API Snowpark - s'exécute dans Snowflake (pas en local)
df = session.table("raw.orders")

result = (
    df.filter(col("status") == "completed")
    .group_by("region")
    .agg(
        sum_("amount").alias("total_revenue"),
        count("*").alias("order_count")
    )
    .sort("total_revenue", ascending=False)
)

result.write.save_as_table("analytics.revenue_by_region", mode="overwrite")

Exécution dans Snowflake — le DataFrame Snowpark se traduit en SQL qui s'exécute dans le warehouse. Les données ne sortent pas
UDFs Python — déployer des fonctions Python dans Snowflake, appelables en SQL depuis n'importe quel outil
Snowpark ML — entraîner des modèles sklearn/XGBoost directement dans Snowflake avec Snowpark

6Optimisation des coûts Snowflake

Question discriminante

Quelles sont vos pratiques pour optimiser les coûts Snowflake en production ?

Auto-suspend obligatoire — 1-2 minutes pour les warehouses interactifs, 5 minutes pour les ETL. Ne jamais laisser un warehouse sans auto-suspend
Warehouse sizing — commencer XS, scaler uniquement si les requêtes sont lentes. Un warehouse M consomme 8x plus qu'un XS pour la même requête simple
Clustering keys — sur les grandes tables (>100GB) filtrées systématiquement sur les mêmes colonnes (date, region)
Result cache — résultats mis en cache 24h si la table n'a pas changé. Standardiser les requêtes BI pour exploiter le cache
Resource monitors — alertes et suspensions automatiques quand le budget mensuel de credits est atteint

-- Resource monitor : alerter à 80% et suspendre à 100% du budget mensuel
CREATE OR REPLACE RESOURCE MONITOR monthly_budget
    WITH CREDIT_QUOTA = 500  -- 500 crédits/mois
    TRIGGERS
        ON 80 PERCENT DO NOTIFY
        ON 100 PERCENT DO SUSPEND_IMMEDIATE;

ALTER WAREHOUSE ETL_WH SET RESOURCE_MONITOR = monthly_budget;

7Grille par niveau

Niveau	Maîtrise	Signal GO	NO-GO
Confirmé	Streams basiques, Tasks, Time Travel, Zero-Copy Clone	A configuré un Stream + Task pour du CDC, a utilisé Time Travel pour débugger	Ne sait pas ce qu'est un Stream Snowflake
Senior	Dynamic Tables, Snowpark, optimisation coûts, resource monitors	A déployé des Dynamic Tables, utilise Snowpark, a mis en place un resource monitor	N'a jamais touché à Snowpark ni aux Dynamic Tables

1Streams: Native CDC in Snowflake

Discriminating question

What is a Snowflake Stream? What is the difference between APPEND_ONLY and DEFAULT?

-- Create a stream on a source table
CREATE STREAM orders_stream ON TABLE orders
    APPEND_ONLY = FALSE;  -- captures INSERT + UPDATE + DELETE

-- APPEND_ONLY = TRUE : captures INSERT only (more performant)
-- APPEND_ONLY = FALSE : captures INSERT + UPDATE + DELETE via METADATA$ACTION

-- Consume the stream in a MERGE
MERGE INTO orders_target AS t
USING (
    SELECT order_id, customer_id, amount,
           METADATA$ACTION,      -- INSERT or DELETE
           METADATA$ISUPDATE     -- True if it's an UPDATE (represented as DELETE+INSERT)
    FROM orders_stream
) AS s ON t.order_id = s.order_id
WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' AND NOT s.METADATA$ISUPDATE 
    THEN DELETE
WHEN MATCHED AND s.METADATA$ACTION = 'INSERT' AND s.METADATA$ISUPDATE 
    THEN UPDATE SET t.amount = s.amount, t.customer_id = s.customer_id
WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT' 
    THEN INSERT (order_id, customer_id, amount) VALUES (s.order_id, s.customer_id, s.amount);

-- Check if the stream has data to consume
SELECT SYSTEM$STREAM_HAS_DATA('orders_stream');

Streams are pointers — they point to an offset in the table history (like Kafka). Consuming the stream advances the offset
Multiple streams — several streams can point to the same table independently
Expiration — if a stream is not consumed within the Time Travel window (7-90 days), it expires and must be recreated

2Tasks: Native Snowflake Orchestration

Discriminating question

How do you combine Streams and Tasks for an automatic CDC pipeline?

-- Task scheduled every hour
CREATE OR REPLACE TASK refresh_aggregates
    WAREHOUSE = 'ETL_WH'
    SCHEDULE = 'USING CRON 0 * * * * UTC'
    ERROR_INTEGRATION = 'my_notification_int'  -- alert on error
AS
    CALL refresh_daily_revenue();

ALTER TASK refresh_aggregates RESUME;

-- Task automatically triggered by a Stream (when new data arrives)
CREATE OR REPLACE TASK process_new_orders
    WAREHOUSE = 'ETL_WH'
    AFTER refresh_aggregates  -- parent task (task DAG)
    WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
    MERGE INTO orders_processed AS t
    USING orders_stream AS s ON t.order_id = s.order_id
    WHEN MATCHED THEN UPDATE SET t.amount = s.amount
    WHEN NOT MATCHED THEN INSERT VALUES (s.order_id, s.customer_id, s.amount);

-- Monitor tasks
SELECT *
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START => DATEADD('hour', -24, CURRENT_TIMESTAMP)
))
ORDER BY SCHEDULED_TIME DESC;

3Dynamic Tables: Automatic Incremental Refresh

Discriminating question

What is a Dynamic Table? When should you use it instead of a dbt model?

-- Create a Dynamic Table
CREATE OR REPLACE DYNAMIC TABLE analytics.daily_revenue
    TARGET_LAG = '1 hour'  -- desired freshness
    WAREHOUSE = ETL_WH
AS
SELECT
    DATE(order_date) AS day,
    region,
    SUM(amount) AS revenue,
    COUNT(*) AS order_count
FROM raw.orders
GROUP BY 1, 2;

-- Snowflake automatically manages incremental refresh
-- No external scheduler or manual incremental logic needed

-- Check status
SHOW DYNAMIC TABLES LIKE 'daily_revenue';

TARGET_LAG — acceptable delay between sources and the Dynamic Table. '1 hour' = data fresher than 1h
vs Materialized Views — Dynamic Tables support complex JOINs, CTEs, subqueries. Much more expressive
vs dbt incremental — Dynamic Tables are automatic (no scheduler needed). dbt incremental requires external orchestration
Cost — Snowflake charges for refresh compute. A short TARGET_LAG = more refreshes = higher cost

4Zero-Copy Cloning

Discriminating question

What is Zero-Copy Cloning? How do you use it in practice?

-- Clone an entire database in seconds, without copying data
CREATE DATABASE staging_clone CLONE production;
-- Data is NOT duplicated. Micro-partitions are shared.
-- Writing to staging → only modified partitions are copied (copy-on-write)

-- Use cases:
-- 1. Risk-free migration testing
-- 2. Dev environment with production data
-- 3. Quick snapshot before a risky operation

-- Time Travel: query past data
SELECT * FROM orders AT(TIMESTAMP => '2025-01-01 00:00:00'::TIMESTAMP_TZ);
SELECT * FROM orders BEFORE(STATEMENT => 'last_bad_query_id');

-- Restore a table from Time Travel
CREATE TABLE orders_restored CLONE orders
    AT(TIMESTAMP => '2025-01-01 00:00:00'::TIMESTAMP_TZ);

-- Undrop: restore a dropped table (within the Time Travel window)
DROP TABLE orders;
UNDROP TABLE orders;  -- works for 7 to 90 days depending on config

5Snowpark Python: Python Inside Snowflake

Discriminating question

What is Snowpark? When should you use it instead of an external Python script?

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, when

session = Session.builder.configs({
    "account": "my-account",
    "user": "data_engineer",
    "private_key": load_private_key()
}).create()

# Snowpark DataFrame API - runs inside Snowflake (not locally)
df = session.table("raw.orders")

result = (
    df.filter(col("status") == "completed")
    .group_by("region")
    .agg(
        sum_("amount").alias("total_revenue"),
        count("*").alias("order_count")
    )
    .sort("total_revenue", ascending=False)
)

result.write.save_as_table("analytics.revenue_by_region", mode="overwrite")

Execution inside Snowflake — the Snowpark DataFrame is translated into SQL that runs in the warehouse. Data never leaves
Python UDFs — deploy Python functions inside Snowflake, callable in SQL from any tool
Snowpark ML — train sklearn/XGBoost models directly inside Snowflake using Snowpark

6Snowflake Cost Optimization

Discriminating question

What are your practices for optimizing Snowflake costs in production?

Mandatory auto-suspend — 1-2 minutes for interactive warehouses, 5 minutes for ETL. Never leave a warehouse without auto-suspend
Warehouse sizing — start XS, scale up only if queries are slow. An M warehouse consumes 8x more than an XS for the same simple query
Clustering keys — on large tables (>100GB) systematically filtered on the same columns (date, region)
Result cache — results cached for 24h if the table has not changed. Standardize BI queries to leverage the cache
Resource monitors — automatic alerts and suspensions when the monthly credit budget is reached

-- Resource monitor: alert at 80% and suspend at 100% of monthly budget
CREATE OR REPLACE RESOURCE MONITOR monthly_budget
    WITH CREDIT_QUOTA = 500  -- 500 credits/month
    TRIGGERS
        ON 80 PERCENT DO NOTIFY
        ON 100 PERCENT DO SUSPEND_IMMEDIATE;

ALTER WAREHOUSE ETL_WH SET RESOURCE_MONITOR = monthly_budget;

7Level Grid

Level	Mastery	GO Signal	NO-GO
Mid-level	Basic Streams, Tasks, Time Travel, Zero-Copy Clone	Has configured a Stream + Task for CDC, has used Time Travel for debugging	Does not know what a Snowflake Stream is
Senior	Dynamic Tables, Snowpark, cost optimization, resource monitors	Has deployed Dynamic Tables, uses Snowpark, has set up a resource monitor	Has never worked with Snowpark or Dynamic Tables

Vous recrutez un profil data ?

Premier entretien gratuit. Rapport GO/NO-GO sous 48h.

Tester gratuitement Reserver un appel