Creare pipeline di dati in Snowflake: una guida per principianti

Le pipeline di dati sono un componente centrale dell’ingegneria dei dati. Rappresentano la serie di passaggi che spostano i dati grezzi dai sistemi sorgente a un data warehouse, dove possono essere ripuliti, trasformati e resi pronti per l’analisi. Una pipeline ben progettata garantisce che i dati fluiscano in modo efficiente, affidabile e sicuro dall’ingestione all’insight.

Snowflake, una potente piattaforma dati basata sul cloud, è particolarmente adatta per costruire pipeline di dati scalabili, flessibili e a costi contenuti. Con funzionalità integrate per il caricamento, la trasformazione, l’automazione e il monitoraggio dei dati, Snowflake permette ai team dati di semplificare flussi di lavoro complessi e concentrarsi sulla generazione di valore dai dati.

In questo tutorial, vedremo il processo di creazione di una pipeline in Snowflake—dalla configurazione dell’ambiente all’automazione delle trasformazioni usando strumenti nativi come Streams e Tasks. Alcuni passaggi richiedono conoscenze sull’integrazione con un servizio cloud e sui sistemi di gestione degli eventi. Per questi, ti consiglio la documentazione del tuo specifico provider cloud.

Se stai iniziando ora con Snowflake, valuta il nostro percorso di competenze Snowflake Foundations per fare pratica e imparare a interrogare, modellare e gestire i dati. 

Comprendere l’architettura di Snowflake

La particolare architettura di Snowflake è ben progettata per supportare le pipeline di dati. Snowflake separa compute e storage, il che significa che puoi scalarli in modo indipendente. 

I dati sono archiviati centralmente nel livello di storage cloud, mentre l’elaborazione avviene nei virtual warehouse, che possono essere avviati, arrestati e ridimensionati secondo necessità.

Questa architettura consente a più utenti e processi di accedere agli stessi dati senza contesa, rendendola ideale per carichi di lavoro di pipeline che prevedono ingestione, trasformazione e analisi. 

Aggiunge inoltre un livello di sicurezza, grazie al quale le modifiche ai dati possono avvenire senza influire direttamente sulla possibilità degli utenti di interrogarli. 

Per ulteriori informazioni e per conoscere il livello cloud non trattato qui, ti consiglio questo approfondimento sull’architettura di Snowflake.

Componenti chiave per le pipeline in Snowflake

All’interno di ogni pipeline dobbiamo considerare i passaggi necessari. In genere, si effettua lo staging dei dati prima di caricarli in Snowflake. Snowflake gestisce lo storage dei dati, incluse le tabelle e i relativi schemi, mentre i virtual warehouse forniscono le risorse di calcolo necessarie per eseguire query e altre operazioni sui dati.

Ecco definizioni chiare dei componenti essenziali usati nelle pipeline di Snowflake:

• Stages: posizioni di storage temporanee o esterne utilizzate per il caricamento/scaricamento dei dati.
• Virtual Warehouses: si riferiscono specificamente alle risorse di calcolo fornite da Snowflake. I virtual warehouse sono essenzialmente cluster di calcolo MPP (Massively Parallel Processing), composti da più macchine virtuali (VM). Snowflake effettua il provisioning e gestisce queste risorse per elaborare query ed eseguire trasformazioni in modo efficiente.
• Databases: contenitori logici per organizzare i tuoi dati. Spesso contengono una raccolta di tabelle.
• Schemas: suddivisioni all’interno di un database per raggruppare tabelle e oggetti correlati. Definiscono come le tabelle sono tra loro collegate.
• Tables: dove vengono archiviati i dati strutturati.

Configurare il tuo ambiente Snowflake

Assicuriamoci che tu abbia un ambiente Snowflake pronto all’uso. Se sei nuovo su Snowflake, segui questi passaggi per iniziare:

Crea un account Snowflake

Se non hai un account Snowflake, è disponibile una prova gratuita di 30 giorni che include (al momento della scrittura) 400 $ di utilizzo gratuito. Dovrebbe essere sufficiente per testare alcune pipeline di esempio!

1. Visita Snowflake Trial e registrati per un account gratuito.

Modulo di registrazione per la prova di Snowflake

2. Per i nostri scopi, selezioneremo l’edizione Standard anziché l’Enterprise per risparmiare sui crediti in futuro se continuerai a usare Snowflake!
3. Scegli il tuo provider cloud preferito (AWS, Azure o GCP). Gli esempi in questo tutorial useranno AWS.
4. Il data center selezionato automaticamente dovrebbe essere quello geograficamente più vicino a te e offrire la migliore esperienza.
5. Dopo la creazione dell’account, accedi alla Web UI di Snowflake.

Navigare nell’interfaccia

Una volta dentro, vedrai:

• Worksheet: dove puoi eseguire query SQL.
• Databases: visualizza e gestisci database e schemi.
• Warehouses: monitora e controlla le risorse di calcolo.
• History: traccia le query e le attività eseguite in precedenza

Se vuoi altri consigli su come usare Snowflake, segui questo corso Snowflake Foundations.

Creazione delle risorse necessarie

Questo passaggio copre la creazione di un virtual warehouse, un database e uno schema. Esegui il seguente SQL nel tuo worksheet per configurare l’ambiente:

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

Caricamento dei dati in Snowflake

Come discutiamo nel nostro articolo sull’ingestione dei dati in Snowflake, Snowflake supporta sia il caricamento batch (bulk) sia quello continuo. Il bulk loading è progettato per caricare grandi blocchi di dati nella tabella in un’unica soluzione. In genere avviene a intervalli pianificati o viene automatizzato con la gestione basata su eventi di Snowpipe. 

Il caricamento continuo (stream) può essere ottenuto usando strumenti come Snowpipe Streaming. Per ingerire dati, dobbiamo creare integrazioni (stages) con il nostro provider cloud e caricare i dati usando COPY INTO.

Creazione dello stage e dell’integrazione S3

L’ingestione dei dati è il primo passo in qualsiasi pipeline. In questo passaggio, vedremo le basi per creare un’integrazione di storage e uno stage. Dovremmo immaginare di aver già connesso Snowflake ad AWS con un utente IAM di Snowflake. In caso contrario, fai riferimento alla guida di configurazione AWS per Snowflake nella documentazione.

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

Alla fine, dovresti aver creato con successo l’integrazione di Snowflake con AWS e lo stage. Se vuoi ripassare le basi di AWS, il seguente articolo è un’ottima introduzione ad AWS.

Bulk loading con COPY INTO

Il comando COPY INTO è il metodo comune per caricare dati su Snowflake. Possiamo usare un file caricato localmente in Snowflake o le nostre connessioni cloud. Nel seguente esempio, useremo il nostro stage per copiare un file 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');

Automatizzare il caricamento con Snowpipe

Caricare manualmente ogni volta che arriva un nuovo file è inefficiente e richiede tempo man mano che la velocità aumenta. Invece, possiamo usare Snowpipe per caricare i file dal nostro stage. Poi, colleghiamo le nostre notifiche di evento AWS al nostro utente IAM di Snowflake per attivare Snowpipe a ogni arrivo.

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

Successivamente, possiamo usare qualcosa come Amazon SQS per inviare notifiche al nostro utente IAM ogni volta che il bucket di destinazione riceve un file.

Infine, possiamo monitorare Snowpipe usando:

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;

Ora non devi più preoccuparti di caricare i dati ogni volta che un file arriva nel tuo bucket S3! Queste notifiche cloud attiveranno Snowpipe per il caricamento.

Trasformare i dati con Streams e Tasks

Streams e Tasks di Snowflake sono strumenti integrati per tracciare e trasformare i dati in cambiamento. Gli Streams si concentrano sul tracciamento delle modifiche alle tabelle, mentre i Tasks vengono utilizzati per automatizzare l’esecuzione di istruzioni SQL.

Introduzione agli Streams

Uno Stream cattura le informazioni di Change Data Capture (CDC)—cioè inserimenti, aggiornamenti ed eliminazioni—su una tabella a partire dall’ultima volta in cui è stato interrogato. Spesso si tratta di query di tipo DML (Data Manipulation Language). Questo ci consente di vedere le modifiche a una tabella tra due momenti di interrogazione diversi.

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

Puoi quindi interrogare lo stream per ottenere solo le modifiche:

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

Introduzione ai Tasks

I Tasks ti permettono di automatizzare l’esecuzione di SQL su una pianificazione o al verificarsi di un evento. I Tasks possono richiamare SQL, Javascript, Python e così via, per automatizzare procedure sulle nostre tabelle e pipeline.

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

Combinare Streams e Tasks

Questa è la base delle pipeline di trasformazione automatizzate. Come hai visto sopra, i Tasks possono usare gli Streams per prelevare nuovi dati in un’altra tabella. Un processo generale può assomigliare al seguente:

• Creare uno Stream per monitorare le modifiche a una tabella di dati
• Creare un Task che preleva i dati da quello stream su base pianificata
• A ogni prelievo, trasformare i dati nel formato desiderato
• Caricare quei dati nella tabella di destinazione

In questo modo, abbiamo creato parti di una pipeline di dati. Combinando Streams e Tasks, possiamo eseguire trasformazioni solo sui dati appena caricati. Con i soli Tasks, potremmo dover elaborare ogni volta l’interezza delle nostre tabelle raw per ottenere nuovi dati: questo può essere costoso e un uso inefficiente della potenza di calcolo.

Costruire una pipeline di dati end-to-end

Vediamo un semplice caso d’uso di pipeline: ingestione dei dati degli ordini, tracciamento delle modifiche e trasformazione per l’analisi. Poiché abbiamo trattato come eseguire ogni passaggio singolarmente, questa sezione mostra cosa potrebbero essere i passaggi nel loro insieme. 

Passaggio 1: crea l’integrazione con lo storage cloud e lo stage

Per prima cosa, dobbiamo collegare l’ecosistema Snowflake al nostro storage cloud. Abbiamo usato AWS come esempio. In Snowflake, abbiamo utilizzato i comandi CREATE STORAGE INTEGRATION e CREATE STAGE per creare la connessione tra AWS e il nostro 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;

Passaggio 2: ingerisci i dati con Snowpipe

In questo secondo passaggio, creiamo uno Snowpipe che eseguirà il comando COPY INTO nelle nostre tabelle di dati raw. 

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

Passaggio 3: usa gli Streams per catturare le modifiche

Poi usiamo uno stream su questa tabella di dati raw per tracciare le modifiche.

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

Passaggio 4: automatizza le trasformazioni con i Tasks

Successivamente, creiamo un Task che trasforma e carica questi dati appena arrivati nella nostra versione “processed” della tabella. Questo verrà attivato secondo una pianificazione per verificare la presenza di nuovi dati. Se non ci sono nuovi dati, il Task non verrà eseguito. 

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

Passaggio 5: monitora e gestisci

Infine, vogliamo assicurarci che le nostre pipeline funzionino come previsto. Ricorda di usare i comandi SHOW PIPES e SHOW TASKS per verificare che siano configurati correttamente e in esecuzione. Poi usa le tabelle di metadati di INFORMATION_SCHEMA per monitorare le prestazioni di caricamento dei dati e gli errori.

Best practice per lo sviluppo di pipeline dati in Snowflake

Come sempre, vogliamo assicurarci di operare secondo le best practice durante lo sviluppo. Gli aspetti da considerare includono l’ottimizzazione di prestazioni e costi, la garanzia di qualità e affidabilità dei dati e l’implementazione dei giusti livelli di sicurezza e accesso.

Ottimizza per prestazioni e costi

Il nostro obiettivo dovrebbe essere far funzionare sempre le pipeline nel modo più efficiente possibile con il minor costo. In questo modo, riduciamo l’impatto sui server e forniamo i dati nel modo più tempestivo possibile. Ecco alcuni suggerimenti: 

• Usa l’auto-suspend sui warehouse per evitare addebiti a vuoto. Questo riduce i costi di compute dovuti ai warehouse in esecuzione, poiché la tariffazione è a tempo.
• Carica file compressi e raggruppali in dimensioni ottimali. File troppo piccoli comportano un eccesso di overhead di rete per l’apertura di molte connessioni. File troppo grandi possono saturare la banda e rischiare il fallimento. Snowflake raccomanda delta di 100–250 MB.
• Pianifica le trasformazioni nelle finestre di basso utilizzo. Così non competi per le risorse.

Garantisci qualità e affidabilità dei dati

Qualità e affidabilità dei dati sono fondamentali per ogni data engineer. Vuoi assicurarti che i dati caricati siano utilizzabili senza errori. Vuoi anche che la tua pipeline non causi involontariamente perdita di dati in caso di problemi. Ecco alcuni suggerimenti per la gestione della qualità dei dati:

• Valida i dati prima e dopo l’ingestione. Può essere fatto con controlli periodici delle tabelle di staging e con alert.
• Usa streams e tabelle di staging per isolare e correggere i record problematici. Poiché vengono prelevati solo i dati modificati, possiamo ridurre al minimo la portata dei problemi usando gli stream.
• Applica vincoli di unicità o logiche di deduplicazione dove necessario. Questo può prevenire, ad esempio, dati NULL dove non sono ammessi o permettere la deduplicazione anticipata in caso di pull sovrapposti.

Implementa sicurezza e controlli di accesso

Infine, è importante garantire il corretto accesso e la sicurezza dei dati. Alcune tabelle possono contenere informazioni sensibili o non dovrebbero essere modificate se non da un gruppo selezionato. Segui questi consigli per gestire in modo corretto sicurezza e accesso in Snowflake:

• Usa ruoli e concessioni per limitare l’accesso.
• Proteggi i dati sensibili con mascheramento a livello di colonna o policy di accesso a livello di riga.
• Ruota le credenziali di storage e monitora l’utilizzo degli stage.

Conclusione

Creare pipeline di dati in Snowflake è un processo snello e scalabile, perfetto per chi muove i primi passi nel mondo del data engineering. Con la sua architettura intuitiva, le robuste opzioni di ingestione e strumenti di automazione integrati come Streams e Tasks, Snowflake consente ai team dati di passare rapidamente dai dati grezzi a insight azionabili.

Seguendo questa guida, ora hai le conoscenze di base per impostare la tua pipeline in Snowflake—gestendo ingestione, trasformazione e orchestrazione, tutto all’interno di un’unica piattaforma. Per maggiori informazioni su Snowflake, consulta le seguenti guide:

• Snowflake per principianti
• Corso Introduction to Snowflake
• Come imparare Snowflake
• Data Modeling in Snowflake
• Migliori certificazioni Snowflake 2025