Construire des pipelines de données avec Snowflake : Un guide pour les débutants

Les pipelines de données sont un élément central de l'ingénierie des données. Ils représentent la série d'étapes qui permettent de transférer les données brutes des systèmes sources vers un entrepôt de données, où elles peuvent être nettoyées, transformées et préparées pour l'analyse. Un pipeline bien conçu garantit que les données circulent de manière efficace, fiable et sécurisée, de l'ingestion à l'analyse.

Snowflake, une puissante plateforme de données basée sur le cloud, est particulièrement bien adaptée à la construction de pipelines de données évolutifs, flexibles et rentables. Avec des fonctionnalités intégrées pour le chargement, la transformation, l'automatisation et la surveillance des données, Snowflake permet aux équipes chargées des données de simplifier les flux de travail complexes et de se concentrer sur la création de valeur à partir des données.

Dans ce tutoriel, nous allons parcourir le processus de construction d'un pipeline de données dans Snowflake - de la configuration de votre environnement à l'automatisation des transformations de données à l'aide d'outils natifs tels que Streams et Tasks. Pour cela, il faudra en partie connaître le processus d'intégration d'un service cloud et les systèmes de gestion des événements. Pour cela, je vous recommande de consulter la documentation de votre service cloud particulier.

Si vous débutez dans l'utilisation de Snowflake, envisagez notre cursus de compétences Snowflake Foundations pour mettre la main à la pâte et apprendre à interroger, modéliser et gérer les données.

Comprendre l'architecture de Snowflake

L'architecture unique de Snowflake est bien conçue pour prendre en charge les pipelines de données. Snowflake sépare le calcul et le stockage, ce qui signifie que vous pouvez faire évoluer chacun d'entre eux indépendamment. 

Les données sont stockées de manière centralisée dans la couche de stockage du cloud, tandis que le calcul s'effectue dans des entrepôts virtuels, qui peuvent être démarrés, arrêtés et redimensionnés selon les besoins.

Cette architecture permet à plusieurs utilisateurs et processus d'accéder aux mêmes données sans conflit, ce qui la rend idéale pour les charges de travail en pipeline qui impliquent l'ingestion, la transformation et l'analyse. 

Il ajoute également une couche de sécurité qui permet de modifier les données sans affecter directement la capacité d'un utilisateur à interroger ces données. 

Pour plus d'informations et pour en savoir plus sur la couche cloud qui n'est pas abordée ici, je vous recommande cette plongée dans l'architecture de Snowflake.

Composants clés des pipelines Snowflake

Pour chaque filière, nous devons envisager les étapes nécessaires. En général, vous mettez en scène les données avant de les stocker dans un entrepôt virtuel. Cet entrepôt virtuel contiendra la base de données qui stocke les tableaux et leur schéma. Voici les définitions de ces éléments essentiels utilisés dans les pipelines de données Snowflake :

• Étapes: Emplacements de stockage temporaires ou externes utilisés pour charger/décharger des données.
• Entrepôts virtuels: Ressources informatiques utilisées pour traiter les requêtes SQL et les opérations sur les données.
• Bases de données: Conteneurs logiques pour organiser vos données. Ceux-ci stockent souvent un ensemble de tableaux
• Schémas: Subdivisions au sein d'une base de données permettant de regrouper des tableaux et des objets apparentés. Elle définit la manière dont les tableaux sont liés les uns aux autres.
• Tableaux: Où sont stockées les données structurées

Configuration de votre environnement Snowflake

Assurez-vous que votre environnement Snowflake est prêt à fonctionner. Si vous ne connaissez pas encore Snowflake, suivez les étapes suivantes pour commencer :

Créez un compte Snowflake

Si vous n'avez pas de compte Snowflake, vous pouvez bénéficier d'une période d'essai gratuite de 30 jours qui inclut (au moment de la rédaction du présent document) 400 dollars d'utilisation gratuite. Cela devrait suffire pour tester quelques exemples de pipelines !

1. Visitez le site Essai de Snowflake et ouvrez un compte gratuit.

Formulaire d'inscription pour le Snowflake Trial

2. Dans notre cas, nous choisirons l'édition Standard au lieu de l'édition Enterprise afin d'économiser sur les coûts de crédit à l'avenir si vous continuez à utiliser Snowflake !
3. Choisissez votre fournisseur de cloud préféré (AWS, Azure ou GCP). Les exemples de ce tutoriel utilisent AWS.
4. Le centre de données choisi automatiquement doit être le plus proche de vous géographiquement et doit vous offrir la meilleure expérience.
5. Après la création du compte, connectez-vous à l'interface Web de Snowflake.

Naviguer dans l'interface utilisateur

Une fois à l'intérieur, vous verrez :

• Feuille de travail: Où vous pouvez exécuter des requêtes SQL.
• Bases de données: Visualiser et gérer les bases de données et les schémas.
• Entrepôts: Surveillez et contrôlez les ressources informatiques.
• Histoire: Cursus des requêtes et des tâches précédemment exécutées

Si vous voulez plus de conseils sur l'utilisation de Snowflake, suivez ceci Fondements de Snowflake cours.

Créer les ressources nécessaires

Cette étape couvre la création d'un entrepôt virtuel, d'une base de données et d'un schéma. Exécutez le code SQL suivant dans votre feuille de calcul pour configurer votre environnement :

-- set the user role appropriately
USE ROLE accountadmin;

-- Create a virtual warehouse
CREATE WAREHOUSE datacamp_pipeline_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;

-- Set our active working warehouse
USE WAREHOUSE datacamp_pipeline_wh;

-- create our Datacamp Sample Database
CREATE OR REPLACE DATABASE datacamp_sample_db;

-- create the Datacamp (Sample) Raw Schema
CREATE OR REPLACE SCHEMA datacamp_sample_db.raw_schema;

Chargement des données dans Snowflake

Comme nous l'expliquons dans notre article sur L'ingestion de données dans SnowflakeSnowflake supporte à la fois le chargement par lots (bulk) et le chargement continu. Le chargement en masse est conçu pour télécharger en une seule fois de grandes quantités de données dans votre tableau. Cette opération est généralement effectuée à intervalles réguliers ou automatisée grâce à la gestion par événement de Snowpipe. 

Le chargement continu (en flux) peut être réalisé à l'aide d'outils tels que Snowpipe Stream. Pour ingérer des données, nous devons créer des intégrations de données (étapes) avec notre fournisseur de cloud et charger les données à l'aide de COPY INTO.

Création de la scène S3 et intégration

L'ingestion des données est la première étape de tout pipeline. Dans cette étape, nous allons couvrir les bases de la création d'une intégration de stockage et de la création d'une étape. Pour ce faire, il faut imaginer que nous avons déjà connecté votre Snowflake à AWS avec un utilisateur IAM Snowflake. Si ce n'est pas le cas, consultez le Guide de configuration de Snowflake AWS dans la documentation.

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
-- The STORAGE_AWS_ROLE_ARN will come from the AWS configuration guide steps
-- We will allow access to all locations and block any with sensitive data
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;

À la fin, vous devriez avoir réussi à créer votre intégration Snowflake avec AWS et à créer la scène. Si vous souhaitez revoir les bases d'AWS, l'article suivant est une excellente introduction à AWS.

Chargement en masse avec COPY INTO

La commande COPY INTO est la méthode la plus courante pour télécharger des données vers Snowflake. Nous pouvons soit utiliser un fichier téléchargé localement dans Snowflake, soit utiliser nos connexions au cloud. Dans l'exemple suivant, nous utiliserons notre étape pour copier un fichier CSV.

-- Load data using COPY INTO
-- The following will attempt to load all file in the bucket
-- To specify the particular files you can use the FILES or PATTERN settings
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @my_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');

Automatiser le chargement avec Snowpipe

Le chargement manuel à chaque fois que nous recevons un nouveau fichier est inefficace et prend du temps lorsque la vitesse augmente. A la place, nous pouvons utiliser Snowpipe pour charger des fichiers depuis notre scène. Ensuite, nous combinons nos notifications d'événements AWS à notre utilisateur IAM Snowflake pour déclencher le Snowpipe à chaque fois.

-- Let’ s first create our Snowpipe
CREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');

Ensuite, nous pouvons utiliser quelque chose comme Amazon SQS pour envoyer des notifications à notre utilisateur IAM chaque fois que notre panier cible reçoit un fichier.

Enfin, nous pouvons surveiller notre Snowpipe en utilisant :

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;

Désormais, vous n'avez plus à vous soucier de télécharger des données à chaque fois qu'un fichier arrive dans votre panier S3 ! Ces notifications dans le cloud déclencheront le chargement des données dans votre Snowpipe. 

Transformer des données avec des flux et des tâches

Les cursus et les tâches de Snowflake sont des outils intégrés qui permettent de suivre et de transformer les données changeantes. Les cursus se concentrent sur le suivi des modifications apportées aux tableaux, tandis que les tâches sont utilisées pour automatiser l'exécution des instructions SQL.

Introduction aux flux

Un flux capture les informations de capture des données de modification (CDC), c'est-à-dire les insertions, les mises à jour et les suppressions effectuées dans un tableau depuis la dernière fois qu'il a été interrogé. Celles-ci se présentent souvent sous la forme de requêtes en langage de manipulation de données (DML). Cela nous permet de voir les changements apportés à un tableau entre deux heures d'interrogation différentes.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;

Vous pouvez ensuite interroger le flux pour obtenir uniquement les modifications :

SELECT * FROM my_stream WHERE METADATA$ACTION = 'INSERT';

Introduction aux tâches

Les tâches vous permettent d'automatiser l'exécution de SQL selon un calendrier ou un déclencheur d'événement. Les tâches peuvent soit faire appel à SQL, Javascript, Python, et ainsi de suite, afin d'automatiser des procédures sur nos tableaux et pipelines.

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;

Combinaison de flux et de tâches

C'est la base des pipelines de transformation automatisés. Comme vous l'avez vu plus haut, les tâches peuvent utiliser des flux de données afin d'introduire de nouvelles données dans un tableau différent. Un processus général peut se présenter comme suit :

• Créez un flux afin de surveiller les modifications apportées à un tableau de données.
• Créez une tâche qui extrait les données de ce flux de manière programmée.
• Pour chaque extraction de données, transformez les données dans le format souhaité.
• Chargez ces données dans notre tableau cible

Nous avons ainsi créé des éléments d'un pipeline de données. En combinant les flux et les tâches, nous sommes en mesure d'effectuer des transformations uniquement sur les données nouvellement chargées. Avec seulement des tâches, nous devrions potentiellement traiter l'intégralité de nos tableaux bruts à chaque fois que nous voulons de nouvelles données, ce qui peut être coûteux et constituer une utilisation inefficace de la puissance de calcul.

Construire un pipeline de données de bout en bout

Passons en revue un cas d'utilisation simple de pipeline : l'ingestion de données de commande, le cursus des changements et leur transformation à des fins d'analyse. Puisque nous avons vu comment exécuter chaque étape indépendamment des autres, nous verrons dans cette section à quoi elles peuvent ressembler ensemble. 

Étape 1 : Créer une intégration et une étape de stockage dans le cloud.

Tout d'abord, nous devons connecter notre écosystème Snowflake à notre stockage dans le cloud. Nous avons pris l'exemple d'AWS. Dans Snowflake, nous avons utilisé les commandes CREATE STORAGE INTEGRATION et CREATE STAGE pour établir la connexion entre notre AWS et notre base de données.

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;

Étape 2 : Ingérer des données avec Snowpipe

Dans cette deuxième étape, nous créons un Snowpipe qui exécutera la commande COPY INTO dans nos tableaux de données brutes. 

-- Let’ s first create our SnowpipeCREATE PIPE my_pipe AS

COPY INTO datacamp_sample_db.raw_schema.aws_first_upload

FROM @datacamp_s3_stage

FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');

Étape 3 : Utiliser les flux pour capturer les changements

Nous utilisons ensuite un flux sur ce tableau de données brutes afin de suivre les changements.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;

Étape 4 : Automatiser les transformations avec des tâches

Ensuite, nous créons une tâche qui transforme et charge ces nouvelles données dans notre version traitée du tableau. Cette opération se déclenchera à intervalles réguliers pour vérifier la présence de nouvelles données. S'il n'y a pas de nouvelles données, la tâche ne sera pas exécutée. 

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;

Étape 5 : Contrôler et gérer

Enfin, nous voulons nous assurer que nos pipelines fonctionnent comme prévu. Veillez à utiliser les commandes SHOW PIPES et SHOW TASKS pour vérifier qu'elles sont correctement configurées et fonctionnent. Utilisez ensuite les tableaux de métadonnées du site INFORMATION_SCHEMA pour suivre les performances et les erreurs de chargement des données.

Meilleures pratiques pour le développement de pipelines de données dans Snowflake

Comme toujours, nous voulons nous assurer que nous appliquons les meilleures pratiques dans notre développement. Les éléments à prendre en compte sont l'optimisation des performances et des coûts, la garantie de la qualité et de la fiabilité des données, et la mise en œuvre des niveaux de sécurité et d'accès appropriés.

Optimiser les performances et les coûts

Notre objectif doit être de toujours exploiter nos pipelines de la manière la plus efficace possible et à moindre coût. De cette manière, nous minimisons l'impact sur les serveurs et fournissons les données le plus rapidement possible. Voici quelques conseils : 

• Utilisez la suspension automatique pour les entrepôts afin d'éviter les frais d'inactivité. Cela permet de réduire les coûts de calcul liés à l'exploitation des entrepôts, qui sont facturés au temps.
• Chargez des fichiers compressés et regroupez-les dans des tailles optimales. Les fichiers trop petits prennent trop de place sur le réseau en raison de la nécessité de générer des connexions pour charger chaque fichier. Les fichiers trop volumineux peuvent prendre trop de bande passante et risquent d'échouer. Snowflake recommande des deltas d'une taille comprise entre 100 et 250 Mo.
• Transformations du calendrier pendant les fenêtres de faible utilisation. Cela signifie que vous n'êtes pas en concurrence pour les ressources.

Garantir la qualité et la fiabilité des données

La qualité et la fiabilité des données sont des éléments essentiels que tout ingénieur des données doit prendre en compte. Vous voulez vous assurer que les données chargées sont utilisables sans erreur. Vous devez également vous assurer que votre pipeline de données n'entraînera pas par inadvertance la perte de données en cas de problème. Voici quelques conseils pour la gestion de la qualité des données :

• Validez les données avant et après l'ingestion. Cela peut se faire par des contrôles périodiques des tableaux de mise à disposition et des alertes.
• Utilisez les flux et les tableaux d'attente pour isoler et corriger les enregistrements problématiques. Étant donné que seules les données modifiées sont introduites, nous pouvons minimiser la portée de nos problèmes en utilisant les flux.
• Appliquez des contraintes uniques ou une logique de déduplication là où c'est nécessaire. Cela permet d'éviter des situations telles que NULL où il n'y a pas de données ou un pré-dédoublement en cas de chevauchement des données extraites.

Mettre en place des contrôles de sécurité et d'accès

Enfin, il est important de garantir un accès et une sécurité appropriés à vos données. Certains tableaux peuvent contenir des informations sensibles ou ne doivent être modifiés que par un groupe restreint. Suivez ces conseils pour gérer correctement votre sécurité et vos accès dans Snowflake :

• Utilisation rôles et subventions pour restreindre l'accès.
• Protégez les données sensibles avec le masquage au niveau des colonnes ou politiques d'accès aux lignes.
• Faites tourner les références de stockage et surveillez l'utilisation de la scène.

Conclusion

La construction de pipelines de données dans Snowflake est un processus rationalisé et évolutif, parfait pour les débutants qui font un pas dans le monde de l'ingénierie des données. Avec son architecture intuitive, ses options d'ingestion robustes et ses outils d'automatisation intégrés tels que les flux et les tâches, Snowflake permet aux équipes chargées des données de passer rapidement des données brutes à des informations exploitables.

En suivant ce guide, vous disposez désormais des connaissances fondamentales pour mettre en place votre propre pipeline Snowflake gérant l'ingestion, la transformation et l'orchestration, le tout au sein d'une plateforme unique. Pour plus d'informations sur Snowflake, consultez les guides suivants :

• Snowflake pour les débutants
• Introduction au cours Snowflake
• Comment apprendre le Snowflake
• Modélisation des données avec Snowflake
• Meilleure certification Snowflake 2025