Data-pipelines bouwen in Snowflake: een beginnersvriendelijke gids

Data-pipelines zijn een kernonderdeel van data-engineering. Ze vormen de reeks stappen die ruwe data van bronsystemen naar een datawarehouse verplaatsen, waar deze kan worden opgeschoond, getransformeerd en klaargemaakt voor analyse. Een goed ontworpen pipeline zorgt ervoor dat data efficiënt, betrouwbaar en veilig stroomt van ingestie tot inzicht.

Snowflake, een krachtig cloudgebaseerd dataplatfom, is bij uitstek geschikt voor het bouwen van schaalbare, flexibele en kostenefficiënte data-pipelines. Met ingebouwde functies voor data­loading, transformatie, automatisering en monitoring stelt Snowflake datateams in staat om complexe workflows te vereenvoudigen en zich te richten op het leveren van waarde uit data.

In deze tutorial lopen we het proces door van het bouwen van een data-pipeline in Snowflake—van het inrichten van je omgeving tot het automatiseren van datatransformaties met native tools zoals Streams en Tasks. Voor sommige onderdelen heb je kennis nodig van de integratieprocessen en eventmanagementsystemen van een cloudservice. Raadpleeg daarvoor de documentatie van jouw specifieke cloudprovider.

Als je net begint met Snowflake, bekijk dan onze Snowflake Foundations skill track om hands-on te leren query’s te schrijven, te modelleren en data te beheren. 

De architectuur van Snowflake begrijpen

Snowflakes unieke architectuur is uitstekend ontworpen om data-pipelines te ondersteunen. Snowflake scheidt compute en storage, wat betekent dat je elk onafhankelijk kunt schalen. 

Data wordt centraal opgeslagen in de cloudopslaglaag, terwijl berekeningen plaatsvinden in virtuele warehouses, die je naar behoefte kunt starten, stoppen en schalen.

Deze architectuur maakt het mogelijk dat meerdere gebruikers en processen dezelfde data zonder conflicten benaderen, wat ideaal is voor pipeline-workloads die ingestie, transformatie en analyse omvatten. 

Het voegt ook een veiligheidslaag toe waarbij wijzigingen in data kunnen plaatsvinden zonder de mogelijkheid van een gebruiker om die data te bevragen direct te beïnvloeden. 

Voor meer informatie en om te leren over de cloudlaag die hier niet is besproken, raad ik deze deep dive in de architectuur van Snowflake aan.

Belangrijke componenten voor Snowflake-pipelines

Binnen elke pipeline moeten we de noodzakelijke stappen overwegen. Gewoonlijk stage je data voordat je die in Snowflake laadt. Snowflake zelf verzorgt de dataopslag, inclusief tabellen en hun schema, terwijl virtuele warehouses de compute­resources leveren die nodig zijn om query’s en andere data-operaties uit te voeren.

Hier zijn heldere definities van deze essentiële componenten die in Snowflake-data-pipelines worden gebruikt:

• Stages: Tijdelijke of externe opslaglocaties die worden gebruikt voor het laden/uitladen van data.
• Virtual Warehouses: Verwijst specifiek naar compute­resources die door Snowflake worden geleverd. Virtuele warehouses zijn in feite Massively Parallel Processing (MPP) compute­clusters, samengesteld uit meerdere Virtual Machines (VM’s). Snowflake levert en beheert deze resources om query’s te verwerken en datatransformaties efficiënt uit te voeren.
• Databases: Logische containers voor het organiseren van je data. Deze bevatten vaak een verzameling tabellen.
• Schemas: Onderdelen binnen een database om gerelateerde tabellen en objecten te groeperen. Dit definieert hoe de tabellen met elkaar samenhangen.
• Tables: Waar gestructureerde data wordt opgeslagen.

Je Snowflake-omgeving inrichten

Laten we zorgen dat je een Snowflake-omgeving klaar hebt staan. Als je nieuw bent met Snowflake, volg dan deze stappen om te beginnen:

Maak een Snowflake-account aan

Als je nog geen Snowflake-account hebt, is er een gratis proefperiode van 30 dagen die (op het moment van schrijven) $400 aan gratis gebruik omvat. Dat zou genoeg moeten zijn om een paar voorbeeldpipelines te testen!

1. Bezoek Snowflake Trial en meld je aan voor een gratis account.

Aanmeldformulier voor Snowflake Trial

2. Voor onze doeleinden kiezen we de Standard-editie in plaats van de Enterprise-editie om in de toekomst op kredietkosten te besparen als je Snowflake blijft gebruiken!
3. Kies je voorkeurscloudprovider (AWS, Azure of GCP). De voorbeelden in deze tutorial gebruiken AWS.
4. Het automatisch gekozen datacenter zou geografisch het dichtst bij je in de buurt moeten zijn en de beste ervaring moeten bieden.
5. Log na het aanmaken van het account in op de Snowflake Web UI.

Door de UI navigeren

Eenmaal binnen zie je:

• Worksheet: Waar je SQL-query’s kunt uitvoeren.
• Databases: Bekijk en beheer databases en schema’s.
• Warehouses: Monitor en beheer compute­resources.
• History: Volg eerder uitgevoerde query’s en taken

Wil je meer tips over het gebruik van Snowflake, volg dan deze Snowflake Foundations cursus.

Benodigde resources aanmaken

In deze stap maken we een virtueel warehouse, database en schema aan. Voer de volgende SQL uit in je worksheet om je omgeving in te richten:

-- 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;

Data laden in Snowflake

Zoals we bespreken in ons artikel over Data-ingestie in Snowflake, ondersteunt Snowflake zowel batch- (bulk) als continue loading. Bulk loading is bedoeld om grote brokken data in één keer naar je tabel te uploaden. Dit gebeurt doorgaans op geplande momenten, of wordt geautomatiseerd met de eventgestuurde aansturing van Snowpipe. 

Continue (stream) loading kan worden gerealiseerd met bijvoorbeeld Snowpipe Streaming. Om data te ingesteren, moeten we data-integraties (stages) met onze cloudprovider aanmaken en de data laden met COPY INTO.

S3-stage en integratie aanmaken

Data-ingestie is de eerste stap in elke pipeline. In deze stap behandelen we de basis van het aanmaken van een storage-integratie en een stage. Daarvoor gaan we ervan uit dat je Snowflake al hebt gekoppeld aan AWS met een Snowflake IAM-gebruiker. Zo niet, raadpleeg dan de Snowflake AWS-configuratiegids in de documentatie.

-- 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;

Aan het einde zou je je Snowflake-integratie met AWS succesvol moeten hebben aangemaakt en de stage gecreëerd. Als je de AWS-basis wilt herhalen, is het volgende artikel een mooie introductie tot AWS.

Bulk loading met COPY INTO

Het commando COPY INTO is de gebruikelijke methode om data naar Snowflake te uploaden. We kunnen ofwel een lokaal geüpload bestand in Snowflake gebruiken, of onze cloudkoppelingen. In het volgende voorbeeld gebruiken we onze stage om een CSV-bestand te kopiëren.

-- 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');

Laden automatiseren met Snowpipe

Handmatig laden elke keer dat we een nieuw bestand krijgen is inefficiënt en tijdrovend naarmate de snelheid toeneemt. In plaats daarvan kunnen we Snowpipe gebruiken om bestanden van onze stage te laden. Vervolgens koppelen we onze AWS-eventnotificaties aan onze Snowflake IAM-gebruiker om de Snowpipe elke keer te triggeren.

-- 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');

Vervolgens kunnen we iets als Amazon SQS gebruiken om onze IAM-gebruiker te informeren wanneer onze doelbucket een bestand ontvangt.

Tot slot kunnen we onze Snowpipe monitoren met:

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;

Nu hoef je je geen zorgen meer te maken over het uploaden van data telkens wanneer er een bestand in je S3-bucket aankomt! Deze cloudnotificaties triggeren je Snowpipe om data te laden. 

Data transformeren met Streams en Tasks

Snowflakes Streams en Tasks zijn ingebouwde tools voor het bijhouden en transformeren van veranderende data. Streams richten zich op het bijhouden van wijzigingen in tabellen, terwijl Tasks worden gebruikt om de uitvoering van SQL-statements te automatiseren.

Introductie tot Streams

Een Stream legt Change Data Capture (CDC) vast—oftewel inserts, updates en deletes—in een tabel sinds de laatste keer dat deze is gequeried. Dit zijn vaak Data Manipulation Language (DML)-query’s. Zo kunnen we de wijzigingen in een tabel tussen twee verschillende querymomenten zien.

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

Je kunt vervolgens de stream bevragen om alleen de wijzigingen op te halen:

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

Introductie tot Tasks

Tasks stellen je in staat SQL-uitvoering te automatiseren op een schema of eventtrigger. Tasks kunnen een beroep doen op SQL, Javascript, Python, enzovoort, om procedures op onze tabellen en pipelines te automatiseren.

-- 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;

Streams en Tasks combineren

Dit is de basis voor geautomatiseerde transformatiepipelines. Zoals je hierboven zag, kunnen Tasks Streams gebruiken om nieuwe data naar een andere tabel te halen. Een algemeen proces kan er als volgt uitzien:

• Maak een Stream om wijzigingen in een datatabel te monitoren
• Maak een Task die op geplande basis data uit die stream ophaalt
• Transformeer bij elke datapull de data naar het gewenste formaat
• Laad die data in onze doeltabel

Op deze manier hebben we onderdelen van een data-pipeline gecreëerd. Door Streams en Tasks te combineren, kunnen we transformaties uitvoeren op alleen nieuw geladen data. Met alleen Tasks zouden we mogelijk elke keer onze volledige ruwe tabellen moeten verwerken als we nieuwe data willen, wat kostbaar en een inefficiënt gebruik van compute­capaciteit kan zijn.

Een end-to-end data-pipeline bouwen

Laten we een eenvoudig pipeline-usecase doorlopen: orderdata ingesteren, wijzigingen bijhouden en deze transformeren voor analytics. Omdat we hebben behandeld hoe je elke stap afzonderlijk uitvoert, laat deze sectie zien hoe de stappen er samen uit kunnen zien. 

Stap 1: Cloudopslagintegratie en stage aanmaken

Eerst moeten we ons Snowflake-ecosysteem verbinden met onze cloudopslag. We gebruikten AWS als voorbeeld. In Snowflake gebruikten we de commando’s CREATE STORAGE INTEGRATION en CREATE STAGE om de verbinding te leggen tussen onze AWS en onze database.

-- 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;

Stap 2: Data ingesteren met Snowpipe

In deze tweede stap maken we een Snowpipe die het COPY INTO-commando uitvoert naar onze ruwe datatabellen. 

-- 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');

Stap 3: Gebruik Streams om wijzigingen vast te leggen

Vervolgens gebruiken we een stream op deze ruwe datatabel om wijzigingen bij te houden.

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

Stap 4: Transformaties automatiseren met Tasks

Vervolgens maken we een Task die deze nieuw geladen data transformeert en laadt naar onze verwerkte versie van de tabel. Dit wordt op een schema getriggerd om te controleren op nieuwe data. Als er geen nieuwe data is, wordt de Task niet uitgevoerd. 

-- 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;

Stap 5: Monitoren en beheren

Tot slot willen we zeker weten dat onze pipelines werken zoals bedoeld. Gebruik de commando’s SHOW PIPES en SHOW TASKS om te verifiëren dat deze correct zijn ingericht en draaien. Gebruik vervolgens metadatatabellen uit INFORMATION_SCHEMA om de laadtperformance en fouten te volgen.

Best practices voor het ontwikkelen van data-pipelines in Snowflake

Zoals altijd willen we ervoor zorgen dat we volgens best practices ontwikkelen. Denk aan optimalisatie voor performance en kosten, het borgen van datakwaliteit en betrouwbaarheid, en het implementeren van de juiste beveiliging en toegangsrechten.

Optimaliseer voor performance en kosten

Ons doel moet zijn om onze pipelines zo efficiënt mogelijk en met zo laag mogelijke kosten te laten draaien. Zo minimaliseren we de belasting van servers en leveren we data zo tijdig mogelijk. Enkele tips: 

• Gebruik auto-suspend op warehouses om kosten bij inactiviteit te vermijden. Dit vermindert compute­kosten doordat warehouses per tijdseenheid rekenen.
• Laad gecomprimeerde bestanden en batch ze naar optimale groottes. Te kleine bestanden veroorzaken veel netwerkoverhead doordat voor elk bestand een connectie moet worden opgezet. Te grote bestanden kunnen te veel bandbreedte vergen en risico op falen geven. Snowflake raadt deltas van 100–250 MB aan.
• Plan transformaties tijdens uren met lage belasting. Zo concurreer je niet om resources.

Borg datakwaliteit en betrouwbaarheid

Datakwaliteit en betrouwbaarheid zijn cruciaal voor elke data engineer. Je wilt zeker weten dat de geladen data bruikbaar is zonder fouten. Je wilt ook zeker weten dat je data-pipeline niet per ongeluk data laat wegvallen bij problemen. Hier zijn enkele tips voor het beheren van datakwaliteit:

• Valideer data vóór en na ingestie. Dit kan met periodieke controles van stagingtabellen en alerts.
• Gebruik streams en stagingtabellen om problematische records te isoleren en te corrigeren. Omdat alleen gewijzigde data wordt opgehaald, kunnen we met streams de impact van problemen beperken.
• Pas unieke constraints of deduplicatielogica toe waar nodig. Dit kan zaken voorkomen zoals NULL-data waar die niet is toegestaan, of vooraf dedupliceren bij overlappende datapulls.

Beveiliging en toegangscontrole implementeren

Tot slot is het belangrijk om de juiste toegang en beveiliging tot je data te waarborgen. Sommige tabellen kunnen gevoelige informatie bevatten of mogen alleen door een selecte groep worden aangepast. Volg dit advies om je beveiliging en toegang in Snowflake goed te beheren:

• Gebruik rollen en grants om toegang te beperken.
• Bescherm gevoelige data met masking op kolomniveau of row access policies.
• Roteer storage-credentials en monitor het gebruik van stages.

Conclusie

Data-pipelines bouwen in Snowflake is een gestroomlijnd en schaalbaar proces dat perfect is voor beginners die de wereld van data-engineering betreden. Met de intuïtieve architectuur, robuuste ingestieopties en ingebouwde automatiseringstools zoals Streams en Tasks, stelt Snowflake datateams in staat om snel te gaan van ruwe data naar bruikbare inzichten.

Door deze gids te volgen, heb je nu de basiskennis om je eigen Snowflake-pipeline op te zetten—van ingestie en transformatie tot orkestratie, allemaal binnen één platform. Voor meer informatie over Snowflake, bekijk de volgende gidsen:

• Snowflake voor beginners
• Cursus Introductie tot Snowflake
• Hoe leer je Snowflake
• Data­modellering in Snowflake
• Beste Snowflake-certificering 2025