Auf einen Blick: Das Erstellen und Verwenden von Snowflake-Streams ermöglicht eine effiziente Nachverfolgung von Tabellenänderungen für das Datenmanagement. Der Leitfaden behandelt Stream-Aufbau, Tabellenerstellung und Automatisierung von Bestandsupdates.
- Streams wirken wie Trigger und erfassen Datenänderungen zur Verarbeitung.
- Enthält ein Szenario zur Bestandsauffüllung zur Veranschaulichung der Nutzung.
In diesem Tutorial zeigen wir, wie man Streams in Snowflake erstellt und verwendet.
Bäche in Snowflake erklärt
Ein Snowflake-Stream – kurz für Table Stream – verfolgt Änderungen an einer Tabelle. Du kannst Snowflake-Streams nutzen, um:
- Emuliere Trigger in Snowflake (im Gegensatz zu Triggern feuern Streams nicht sofort)
- Sammeln Sie Änderungen in einer Staging Table und aktualisieren Sie eine andere Tabelle basierend auf diesen Änderungen mit gewisser Häufigkeit
Hier erstellen wir ein Beispielszenario: ein Lagerbestandsauffüllsystem. Wenn wir Nachschubaufträge erhalten, müssen wir den Lagerbestand auf Lagerbesitz erhöhen.
Wir führen diese Aufgabe manuell aus. In der tatsächlichen Anwendung möchte man sie als Snowflake-Aufgabe mit einem festen Zeitplan ausführen.
Erstellen Sie die Daten, den Strom und die Tabellen
Um weiterzumachen, erstellen Sie die Tabelle mit Bestellungen und Produkten:
Anwendungsfall im Tutorial
- Bestellungen sind Inventarbewegungen.
- Die Produkte enthalten die vorrätige Menge des Vorrats.
Wenn du mit 25 Gegenständen startest und drei Nachschubbestellungen von 25, 25 und 25 machst, hast du am Ende 100 Artikel zur Hand. Summiere diese drei Bestellungen und addiere 75 zum Startsaldo von 25, um 100 zu erhalten.
Erstellen Sie diese zwei Tabellen:
CREATE TABLE orders
(
customernumber varchar(100) PRIMARY KEY,
ordernumber varchar(100),
comments varchar(200),
orderdate date,
ordertype varchar(10),
shipdate date,
discount number,
quantity int,
productnumber varchar(50)
)
create table products (
productnumber varchar(50) primary key,
movementdate datetime,
quantity number,
movementtype varchar(10));
Füge nun ein Produkt zur Produkttabelle hinzu und gib ihm einen Anfangsbestand von 100 Einheiten auf Lager.
insert into products(productnumber, quantity) values ('EE333', 100);
Erstelle jetzt einen Strom in der Order-Tabelle. Snowflake beginnt, Änderungen an dieser Tabelle zu verfolgen.
CREATE OR REPLACE STREAM orders_STREAM on table orders;
Jetzt erstelle eine Bestellung.
insert into orders (customernumber,ordernumber,comments,orderdate,ordertype,shipdate,discount,quantity,productnumber) values ('855','533','jqplygemaq','2020-10-08','sale','2020-10-18','0.10503143596496034','65','EE333')
Dann fragen Sie die orders_stream-Tabelle ab:
select * from orders_stream
Hier sind die Ergebnisse. Sie sehen, dass wir einen Datensatz hinzugefügt haben.
Aktualisieren Sie den Lagerbestand
Wir haben 65 weitere Artikel in den Bestand aufgenommen, daher müssen wir den Lagerbestand aktualisieren. Verfahren wie folgt.
Starte eine Transaktion mit der Start-Anweisung .
begin;
Führen Sie dann diese Aktualisierungsanweisung aus, die im Grunde:
- Summiert die Bestellungen für jedes Produkt
- Addiert die Summe der Bestellmengen zum ursprünglichen Lagersaldo.
Diese Aktualisierungsausweisung erhält die Produktnummern aus der Tabelle des Auftragsstroms. Das ist die Tabelle, die Snowflake anzeigt, welche Produkte ihr Lagerbestand aktualisieren müssen.
update products
set quantity = onhand
from
(select distinct p.productnumber,
p.quantity as dquantity,
o.quantity ,
p.quantity + o.quantity as onhand
from products p
inner join orders o on
p.productnumber = o.productnumber) as z
where z.productnumber = (select productnumber from orders_stream)
commit:
An diesem Punkt ist die orders_stream Tabelle leer, was passiert, wenn man eine Lesung darauf ausführt.
(Hinweis: Start und commit machen Sie eine Transaktion, was eine logisch zusammenhängende Menge von SQL-Anweisungen ist. Sie sperren die beteiligten Tabellen. Ohne das könnte man eine unpassende Situation haben, wie zum Beispiel einen falschen Lagersaldo, weil eine Transaktion funktioniert hat und die andere nicht.)
Fragen Sie nun orders_stream ab und Sie werden sehen, dass die Tabelle leer ist.
Zusätzliche Ressourcen
Für weitere Tutorials wie dieses entdecken Sie diese Ressourcen:
Schneeflocken-Leitfaden
- Snowflake 101: Einführung in die Snowflake Data Cloud
- Wie man Amazon S3-Daten in Snowflake importiert
- Wie man JSON-Daten in Snowflake abfragt
- CSV-Dateien von S3 in Snowflake laden
- Snowflake: Verwendung von Analytik und statistischen Funktionen
- Snowflake-Fensterfunktionen: Partitionieren nach und ordnen nach
- Schneeflocken-Verzögerungsfunktion und gleitende Durchschnitte
- Snowflake SQL Aggregate Functions & Table Joins
- Benutzerdefinierte Funktionen (UDFs) in Snowflake
- Schneeflockenströme erstellen und nutzen
- Verwendung von gespeicherten Verfahren in Snowflake
- Schneeflocke vs. Redshift: Was ist der Unterschied und wie wählt man?
Diese Beiträge sind meine eigenen und spiegeln nicht unbedingt die Position, Strategien oder Meinung von BMC wider.