Aufbau von Datenpipelines in Snowflake: Ein anfängerfreundlicher Leitfaden

Datenpipelines sind ein zentraler Bestandteil des Data Engineering. Sie stellen die Reihe von Schritten dar, mit denen Rohdaten aus den Quellsystemen in ein Data Warehouse übertragen werden, wo sie bereinigt, umgewandelt und für die Analyse vorbereitet werden können. Eine gut durchdachte Pipeline stellt sicher, dass die Daten von der Erfassung bis zur Auswertung effizient, zuverlässig und sicher fließen.

Snowflake, eine leistungsstarke Cloud-basierte Datenplattform, eignet sich besonders gut für den Aufbau skalierbarer, flexibler und kosteneffizienter Datenpipelines. Mit den integrierten Funktionen zum Laden, Umwandeln, Automatisieren und Überwachen von Daten ermöglicht Snowflake es Datenteams, komplexe Arbeitsabläufe zu vereinfachen und sich auf die Wertschöpfung aus Daten zu konzentrieren.

In diesem Lernprogramm gehen wir den Aufbau einer Datenpipeline in Snowflake durch - von der Einrichtung deiner Umgebung bis hin zur Automatisierung von Datentransformationen mit nativen Tools wie Streams und Tasks. Einige dieser Aufgaben erfordern Kenntnisse über den Integrationsprozess und die Ereignisverwaltungssysteme eines Cloud-Dienstes. Hierfür empfehle ich die Dokumentation deines jeweiligen Cloud-Dienstes.

Wenn du gerade erst mit Snowflake anfängst, solltest du unseren Snowflake Lernpfad "Grundlagen um das Abfragen, Modellieren und Verwalten von Daten praktisch zu erlernen.

Die Architektur von Snowflake verstehen

Die einzigartige Architektur von Snowflake ist so konzipiert, dass sie Datenpipelines unterstützt. Snowflake trennt Rechenleistung und Speicherplatz, so dass du beide unabhängig voneinander skalieren kannst. 

Die Daten werden zentral in der Cloud-Speicherschicht gespeichert, während die Berechnungen in virtuellen Lagern stattfinden, die je nach Bedarf gestartet, gestoppt und in der Größe verändert werden können.

Diese Architektur ermöglicht es mehreren Nutzern und Prozessen, auf dieselben Daten zuzugreifen, ohne dass es zu Konflikten kommt, und ist damit ideal für Pipeline-Workloads, die die Aufnahme, Umwandlung und Analyse von Daten beinhalten. 

Außerdem gibt es eine zusätzliche Sicherheitsebene, auf der Datenänderungen vorgenommen werden können, ohne dass die Abfrage der Daten durch den Nutzer direkt beeinträchtigt wird. 

Für weitere Informationen und um mehr über die hier nicht behandelte Cloud-Schicht zu erfahren, empfehle ich diesen tieferen Einblick in die Architektur von Snowflake.

Schlüsselkomponenten für Snowflake-Pipelines

Innerhalb jeder Pipeline müssen wir die notwendigen Schritte berücksichtigen. In der Regel wirst du die Daten vor der Speicherung in einem virtuellen Lager aufbereiten. Dieses virtuelle Lager enthält die Datenbank, in der die Tabellen und ihr Schema gespeichert sind. Hier sind die Definitionen dieser wesentlichen Elemente, die in Snowflake-Datenpipelines verwendet werden:

• Etappen: Temporäre oder externe Speicherorte, die zum Laden/Entladen von Daten verwendet werden.
• Virtuelle Lagerhäuser: Rechenressourcen, die für die Verarbeitung von SQL-Abfragen und Datenoperationen verwendet werden.
• Datenbanken: Logische Container zum Organisieren deiner Daten. Diese speichern oft eine Sammlung von Tabellen
• Schemas: Unterteilungen innerhalb einer Datenbank, um verwandte Tabellen und Objekte zu gruppieren. Hier wird festgelegt, wie die Tabellen zueinander in Beziehung stehen.
• Tabellen: Wo strukturierte Daten gespeichert werden

Einrichten deiner Snowflake-Umgebung

Wir sorgen dafür, dass deine Snowflake-Umgebung einsatzbereit ist. Wenn du neu bei Snowflake bist, befolge diese Schritte, um loszulegen:

Erstelle ein Snowflake-Konto

Wenn du noch kein Snowflake-Konto hast, kannst du es 30 Tage lang kostenlos ausprobieren und dabei (zum Zeitpunkt der Erstellung dieses Artikels) 400 US-Dollar kostenlos nutzen. Das sollte ausreichen, um ein paar Beispielpipelines zu testen!

1. Besuche Snowflake Testversion und melde dich für ein kostenloses Konto an.

Anmeldeformular für Snowflake Trial

2. Für unsere Zwecke wählen wir die Standard-Edition anstelle der Enterprise-Edition, um in Zukunft Kreditkosten zu sparen, wenn du Snowflake! weiter nutzt.
3. Wähle deinen bevorzugten Cloud-Anbieter (AWS, Azure oder GCP). Die Beispiele in diesem Lernprogramm verwenden AWS.
4. Das automatisch ausgewählte Rechenzentrum sollte dasjenige sein, das dir geografisch am nächsten liegt und die beste Erfahrung bietet.
5. Nachdem du ein Konto erstellt hast, melde dich bei der Snowflake Web UI an.

Navigieren in der Benutzeroberfläche

Einmal drinnen, wirst du sehen:

• Arbeitsblatt: Hier kannst du SQL-Abfragen ausführen.
• Datenbanken: Datenbanken und Schemata anzeigen und verwalten.
• Lagerhäuser: Überwache und kontrolliere Rechenressourcen.
• Geschichte: Lernpfad für bereits ausgeführte Abfragen und Aufgaben

Wenn du weitere Tipps zur Verwendung von Snowflake haben möchtest, dann folge diesem Snowflake-Grundlagen Kurs.

Notwendige Ressourcen schaffen

In diesem Schritt geht es um die Erstellung eines virtuellen Lagers, einer Datenbank und eines Schemas. Führe das folgende SQL in deinem Arbeitsblatt aus, um deine Umgebung einzurichten:

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

Daten in Snowflake laden

Wie wir in unserem Artikel über Data Ingestion in Snowflakebeschrieben haben, unterstützt Snowflake sowohl Batch (Bulk) als auch kontinuierliches Laden. Mit dem Bulk Loading kannst du große Datenmengen auf einmal in deine Tabelle hochladen. Dies geschieht in der Regel in regelmäßigen Abständen oder automatisiert mit dem ereignisbasierten Management von Snowpipe. 

Kontinuierliches (Stream-)Laden kann mit Dingen wie Snowpipe Stream erreicht werden. Um Daten aufzunehmen, müssen wir Datenintegrationen (Stages) mit unserem Cloud-Anbieter erstellen und die Daten mit COPY INTO laden.

Erstellen der S3-Bühne und Integration

Die Dateneingabe ist der erste Schritt in jeder Pipeline. In diesem Schritt behandeln wir die Grundlagen der Speicherintegration und der Erstellung einer Stufe. Dazu sollten wir uns vorstellen, dass wir deine Snowflake bereits mit einem Snowflake IAM-Benutzer mit AWS verbunden haben. Falls nicht, lesen Sie die Snowflake AWS Konfigurationsanleitung in der Dokumentation nach.

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

Am Ende solltest du deine Snowflake-Integration mit AWS erfolgreich erstellt und die Stage erstellt haben. Wenn du die Grundlagen von AWS kennenlernen möchtest, ist der folgende Artikel eine gute Einführung in AWS.

Bulk Loading mit COPY INTO

Der Befehl COPY INTO ist die gängige Methode, um Daten in Snowflake hochzuladen. Wir können entweder eine lokal hochgeladene Datei in Snowflake verwenden oder unsere Cloud-Verbindungen nutzen. Im folgenden Beispiel verwenden wir unsere Bühne, um eine CSV-Datei zu kopieren.

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

Automatisiertes Laden mit Snowpipe

Jedes Mal, wenn wir eine neue Datei erhalten, ist das manuelle Laden ineffizient und zeitaufwändig, da die Geschwindigkeit zunimmt. Stattdessen können wir Snowpipe verwenden, um Dateien von unserer Bühne zu laden. Dann kombinieren wir unsere AWS-Ereignisbenachrichtigungen mit unserem Snowflake IAM-Benutzer, um die Snowpipe jedes Mal auszulösen.

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

Als nächstes können wir etwas verwenden wie Amazon SQS verwenden, um Benachrichtigungen an unseren IAM-Benutzer zu senden, wenn unser Ziel-Bucket eine Datei erhält.

Zum Schluss können wir unsere Snowpipe mit überwachen:

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;

Jetzt musst du dich nicht mehr darum kümmern, jedes Mal Daten hochzuladen, wenn eine Datei in deinem S3-Bucket ankommt! Diese Cloud-Benachrichtigungen veranlassen deine Snowpipe, Daten zu laden. 

Daten mit Streams und Tasks umwandeln

Die Streams und Tasks von Snowflake sind integrierte Werkzeuge zum Verfolgen und Umwandeln von sich ändernden Daten. Streams konzentrieren sich auf die Verfolgung von Änderungen an Tabellen, wobei Tasks zur Automatisierung der Ausführung von SQL-Anweisungen verwendet werden.

Einführung in Streams

Ein Stream erfasst Informationen zur Änderungsdatenerfassung (Change Data Capture, CDC), d.h. Einfügungen, Aktualisierungen und Löschungen in einer Tabelle seit der letzten Abfrage. Diese werden oft in Form von DML-Abfragen (Datenbearbeitung Language) gestellt. So können wir die Änderungen an einer Tabelle zwischen zwei verschiedenen Abfragezeitpunkten sehen.

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

Du kannst dann den Stream abfragen, um nur die Änderungen zu erhalten:

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

Einführung in die Aufgaben

Mit Tasks kannst du die SQL-Ausführung nach einem Zeitplan oder Ereignisauslöser automatisieren. Aufgaben können entweder SQL, Javascript, Python usw. verwenden, um Abläufe in unseren Tabellen und Pipelines zu automatisieren.

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

Kombinieren von Streams und Aufgaben

Dies ist die Grundlage für automatisierte Transformationspipelines. Wie du oben gesehen hast, können Aufgaben Streams nutzen, um neue Daten in eine andere Tabelle zu ziehen. Ein allgemeiner Prozess kann wie folgt aussehen:

• Erstelle einen Stream, um Änderungen an einer Tabelle zu überwachen
• Erstelle einen Task, der regelmäßig Daten aus diesem Stream abruft.
• Transformiere die Daten bei jedem Daten-Pull in das gewünschte Format
• Lade diese Daten in unsere Zieltabelle

Auf diese Weise haben wir Teile einer Datenpipeline erstellt. Durch die Kombination von Streams und Tasks sind wir in der Lage, Transformationen nur für neu geladene Daten durchzuführen. Wenn wir nur Tasks verwenden würden, müssten wir jedes Mal, wenn wir neue Daten benötigen, die gesamten Tabellen verarbeiten, was kostspielig und eine ineffiziente Nutzung der Rechenleistung ist.

Aufbau einer End-to-End-Datenpipeline

Gehen wir einen einfachen Anwendungsfall für die Pipeline durch: Bestelldaten aufnehmen, Änderungen verfolgen und für Analysen aufbereiten. Da wir behandelt haben, wie die einzelnen Schritte unabhängig voneinander ausgeführt werden, geht es in diesem Abschnitt darum, wie die Schritte zusammen aussehen können. 

Schritt 1: Cloud-Speicher-Integration erstellen und inszenieren

Zuerst müssen wir unser Snowflake-Ökosystem mit unserem Cloud-Speicher verbinden. Wir haben AWS als Beispiel verwendet. In Snowflake haben wir die Befehle CREATE STORAGE INTEGRATION und CREATE STAGE verwendet, um die Verbindung zwischen unserer AWS und unserer Datenbank herzustellen.

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

Schritt 2: Daten mit Snowpipe einlesen

In diesem zweiten Schritt erstellen wir eine Snowpipe, die den Befehl COPY INTO für unsere Tabellen mit Rohdaten ausführt. 

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

Schritt 3: Verwende Streams, um Änderungen zu erfassen

Wir verwenden dann einen Stream auf dieser Tabelle mit Rohdaten, um Änderungen zu verfolgen.

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

Schritt 4: Automatisiere Transformationen mit Aufgaben

Als Nächstes erstellen wir einen Task, der die neu geladenen Daten transformiert und in unsere verarbeitete Version der Tabelle lädt. Dies wird nach einem bestimmten Zeitplan ausgelöst, um nach neuen Daten zu suchen. Wenn es keine neuen Daten gibt, wird der Task nicht ausgeführt. 

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

Schritt 5: Überwachen und verwalten

Schließlich wollen wir sicherstellen, dass unsere Pipelines wie vorgesehen funktionieren. Stelle sicher, dass du die Befehle SHOW PIPES und SHOW TASKS verwendest, um zu überprüfen, ob diese richtig eingerichtet sind und funktionieren. Verwende dann die Metadaten-Tabellen von INFORMATION_SCHEMA, um die Leistung und Fehler beim Laden der Daten zu verfolgen.

Best Practices für die Entwicklung von Datenpipelines in Snowflake

Wie immer wollen wir sicherstellen, dass wir bei unserer Entwicklung mit den besten Praktiken arbeiten. Es gilt, die Leistung und die Kosten zu optimieren, die Datenqualität und -zuverlässigkeit zu gewährleisten und ein angemessenes Maß an Sicherheit und Zugang zu implementieren.

Optimiere für Leistung und Kosten

Unser Ziel sollte es sein, unsere Pipelines immer so effizient wie möglich und mit so wenig Kosten wie möglich zu betreiben. Auf diese Weise minimieren wir die Auswirkungen auf die Server und stellen die Daten so zeitnah wie möglich zur Verfügung. Hier sind einige Tipps: 

• Nutze die automatische Aussetzung von Lagern, um Leerlaufgebühren zu vermeiden. Das senkt die Kosten für den Betrieb von Lagern, da sie nach Zeit abrechnen.
• Lade komprimierte Dateien und staple sie in die optimale Größe. Zu kleine Dateien beanspruchen zu viel Netzwerk-Overhead, da zum Laden jeder Datei Verbindungen aufgebaut werden müssen. Zu große Dateien können zu viel Bandbreite beanspruchen und zu Ausfällen führen. Snowflake empfiehlt Deltas mit einer Größe von 100-250 MB.
• Planen Sie Umwandlungen während nutzungsarmer Zeitfenster. Das bedeutet, dass du nicht um Ressourcen konkurrierst.

Sicherstellung der Datenqualität und -zuverlässigkeit

Datenqualität und -zuverlässigkeit sind für jeden Dateningenieur entscheidend. Du willst sicherstellen, dass die geladenen Daten ohne Fehler verwendet werden können. Außerdem musst du sicherstellen, dass deine Datenpipeline nicht versehentlich zu Datenverlusten führt, wenn es Probleme gibt. Hier sind einige Tipps für das Datenqualitätsmanagement:

• Validiere die Daten vor und nach der Aufnahme. Dies kann durch regelmäßige Überprüfungen von Tabellen und Warnmeldungen geschehen.
• Nutze Streams und Staging Tabellen, um problematische Datensätze zu isolieren und zu beheben. Da nur geänderte Daten herangezogen werden, können wir den Umfang unserer Probleme mit Streams minimieren.
• Wenden Sie bei Bedarf eindeutige Beschränkungen oder Deduplizierungslogik an. Das kann Dinge wie NULL Daten, die wir nicht haben können, oder Pre-Dedupe bei sich überschneidenden Datenabfragen verhindern.

Implementierung von Sicherheits- und Zugangskontrollen

Schließlich ist es wichtig, dass du den richtigen Zugang zu deinen Daten und deren Sicherheit gewährleistest. Einige Tabellen können sensible Informationen enthalten oder sollten nur von einer ausgewählten Gruppe geändert werden. Befolge diese Ratschläge, um deine Sicherheit und deinen Zugang in Snowflake richtig zu verwalten:

• Verwenden Sie Rollen und Berechtigungen um den Zugriff einzuschränken.
• Schützen Sie sensible Daten mit Maskierung auf Spaltenebene oder Zeilen-Zugriffsrichtlinien.
• Rotiere die Lagerausweise und überwache die Nutzung der Bühne.

Fazit

Der Aufbau von Datenpipelines in Snowflake ist ein rationalisierter und skalierbarer Prozess, der sich perfekt für Anfänger eignet, die in die Welt des Data Engineering einsteigen wollen. Mit seiner intuitiven Architektur, den robusten Ingestion-Optionen und den integrierten Automatisierungstools wie Streams und Tasks ermöglicht Snowflake es Datenteams, schnell von Rohdaten zu verwertbaren Erkenntnissen zu gelangen.

Wenn du diesen Leitfaden befolgst, verfügst du über das nötige Grundwissen, um deine eigene Snowflake-Pipeline einzurichten, die Ingestion, Transformation und Orchestrierung innerhalb einer einzigen Plattform umfasst. Weitere Informationen über Snowflake findest du in den folgenden Leitfäden:

• Snowflake für Anfänger
• Einführung in den Snowflake-Kurs
• Wie lerne ich Snowflake?
• Datenmodellierung in Snowflake
• Beste Snowflake Zertifizierung 2025