Automatisieren von Snowpipe für Google Cloud Storage¶

Schritt 1: Cloud Storage-Integration in Snowflake erstellen¶

Erstellen Sie mit dem Befehl CREATE STORAGE INTEGRATION eine Integration. Eine Integration ist ein Snowflake-Objekt, das die Authentifizierungsverantwortung für externen Cloudspeicher an eine von Snowflake generierte Entität (d. h. ein Cloud Storage-Dienstkonto) delegiert. Für den Zugriff auf Cloud Storage-Buckets erstellt Snowflake ein Dienstkonto, dem Berechtigungen für den Zugriff auf die Buckets erteilt werden können, in denen sich Ihre Datendateien befinden.

Eine einzelne Speicherintegration kann mehrere externe (d. h. GCS) Speicherintegrationen unterstützen. Die URL in der Stagingbereichsdefinition muss mit den für den Parameter STORAGE_ALLOWED_LOCATIONS angegebenen GCS-Buckets (und optionalen Pfaden) übereinstimmen.

CREATE STORAGE INTEGRATION <integration_name>
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://<bucket>/<path>/', 'gcs://<bucket>/<path>/')
  [ STORAGE_BLOCKED_LOCATIONS = ('gcs://<bucket>/<path>/', 'gcs://<bucket>/<path>/') ]

Wobei:

• integration_name ist der Name der neuen Integration.

• bucket ist der Name eines Cloud Storage-Buckets, in dem Ihre Datendateien gespeichert sind (z. B. mybucket). Die erforderlichen Parameter STORAGE_ALLOWED_LOCATIONS und STORAGE_BLOCKED_LOCATIONS beschränken bzw. blockieren den Zugriff auf diese Buckets, wenn Stagingbereiche, die auf diese Integration verweisen, erstellt oder geändert werden.

• path ist ein optionaler Pfad, mit dem Sie Objekte im Bucket genauer steuern können.

Im folgenden Beispiel wird eine Integration erstellt, die externe Stagingbereiche, die die Integration nutzen, explizit darauf beschränkt, auf einen von zwei Buckets und Pfaden zu verweisen: In einem späteren Schritt werden wir einen externen Stagingbereich erstellen, der auf einen dieser Buckets und Pfade verweist.

Zusätzliche externe Stagingbereiche, die diese Integration ebenfalls verwenden, können auf die zulässigen Buckets und Pfade verweisen:

CREATE STORAGE INTEGRATION gcs_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://mybucket1/path1/', 'gcs://mybucket2/path2/')
  STORAGE_BLOCKED_LOCATIONS = ('gcs://mybucket1/path1/sensitivedata/', 'gcs://mybucket2/path2/sensitivedata/');

Copy

Schritt 2: Cloud Storage-Dienstkonto für Ihr Snowflake-Konto abrufen¶

Führen Sie den Befehl DESCRIBE INTEGRATION aus, um die ID für das Cloud Storage-Dienstkonto abzurufen, das automatisch für Ihr Snowflake-Konto erstellt wurde:

DESC STORAGE INTEGRATION <integration_name>;

Wobei:

• integration_name ist der Name der Integration, die Sie in Schritt 1: Cloud Storage-Integration in Snowflake erstellen (unter diesem Thema) erstellt haben.

Beispiel:

DESC STORAGE INTEGRATION gcs_int;

+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+
| property                    | property_type | property_value                                                              | property_default |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------|
| ENABLED                     | Boolean       | true                                                                        | false            |
| STORAGE_ALLOWED_LOCATIONS   | List          | gcs://mybucket1/path1/,gcs://mybucket2/path2/                               | []               |
| STORAGE_BLOCKED_LOCATIONS   | List          | gcs://mybucket1/path1/sensitivedata/,gcs://mybucket2/path2/sensitivedata/   | []               |
| STORAGE_GCP_SERVICE_ACCOUNT | String        | service-account-id@project1-123456.iam.gserviceaccount.com                  |                  |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+

Copy

Die Eigenschaft STORAGE_GCP_SERVICE_ACCOUNT in der Ausgabe zeigt das Cloud Storage-Dienstkonto an, das für Ihr Snowflake-Konto erstellt wurde (z. B. service-account-id@project1-123456.iam.gserviceaccount.com). Wir stellen ein einziges Cloud Storage-Dienstkonto für Ihr gesamtes Snowflake-Konto bereit. Alle Cloud Storage-Integrationen verwenden dieses Dienstkonto.

Schritt 3: Dienstkontoberechtigungen für den Zugriff auf Bucket-Objekte erteilen¶

Die folgende Schritt-für-Schritt-Anleitung beschreibt, wie Sie IAM-Zugriffsberechtigungen für Snowflake über Ihre Google Cloud Platform Console so konfigurieren, dass Sie einen Cloud Storage-Bucket zum Laden und Entladen von Daten verwenden können:

Voraussetzungen¶

In den Anweisungen unter diesem Thema wird davon ausgegangen, dass die folgenden Elemente erstellt und konfiguriert wurden:

GCP-Konto:

• Pub/Sub-Thema, das Ereignismeldungen vom GCS-Bucket empfängt. Weitere Informationen dazu finden Sie unter Erstellen des Pub/Sub-Themas (unter diesem Thema).

• Abonnement, das Ereignismeldungen vom Pub/Sub-Thema empfängt. Weitere Informationen dazu finden Sie unter Erstellen des Pub/Sub-Abonnements (unter diesem Thema).

Eine Anleitung dazu finden Sie unter Pub/Sub-Dokumentation.

Snowflake:

• Zieltabelle in der Snowflake-Datenbank, in die Sie Daten laden möchten.

$ gsutil notification create -t <topic> -f json gs://<bucket-name> -e OBJECT_FINALIZE

Schritt 1: Benachrichtigungsintegration in Snowflake erstellen¶

Erstellen Sie eine Benachrichtigungsintegration mit dem Befehl CREATE NOTIFICATION INTEGRATION. Die Benachrichtigungsintegration verweist auf Ihr Pub/Sub-Abonnement. Snowflake verknüpft die Benachrichtigungsintegration mit einem GCS-Dienstkonto, das für Ihr Konto erstellt wurde. Snowflake erstellt ein einziges Dienstkonto, auf das von allen GCS-Speicherintegrationen in Ihrem Snowflake-Konto verwiesen wird.

CREATE NOTIFICATION INTEGRATION <integration_name>
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = true
  GCP_PUBSUB_SUBSCRIPTION_NAME = '<subscription_id>';

Wobei:

• integration_name ist der Name der neuen Integration.

• subscription_id ist der Abonnementname, den Sie sich unter Abrufen der Pub/Sub-Abonnement-ID notiert haben.

Beispiel:

CREATE NOTIFICATION INTEGRATION my_notification_int
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = true
  GCP_PUBSUB_SUBSCRIPTION_NAME = 'projects/project-1234/subscriptions/sub2';

Schritt 2: Snowflake Zugriff auf das Pub/Sub-Abonnement gewähren¶

1. Führen Sie den Befehl DESCRIBE INTEGRATION aus, um die Snowflake-Dienstkonto-ID abzurufen:

   DESC NOTIFICATION INTEGRATION <integration_name>;
   

   Copy

   Wobei:

   • integration_name ist der Name der Integration, die Sie in Schritt 1: Benachrichtigungsintegration in Snowflake erstellen erstellt haben.

   Beispiel:

   DESC NOTIFICATION INTEGRATION my_notification_int;
   

   Copy

2. Notieren Sie den Namen des Dienstkontos in der Spalte GCP_PUBSUB_SERVICE_ACCOUNT, die das folgende Format hat:

   <service_account>@<project_id>.iam.gserviceaccount.com
   

   Copy

3. Melden Sie sich bei der Google Cloud Platform Console als Projekteditor an.

4. Wählen Sie im Startseiten-Dashboard die Option Big Data » Pub/Sub » Subscriptions aus.

5. Wählen Sie das Abonnement aus, das für den Zugriff konfiguriert werden soll.

6. Klicken Sie oben rechts auf SHOW INFO PANEL. Der Informationsbereich für das Abonnement wird erweitert.

7. Klicken Sie auf die Schaltfläche ADD PRINCIPAL.

8. Suchen Sie im Feld New principals nach dem Dienstkontonamen, den Sie sich notiert haben.

9. Wählen Sie in der Dropdown-Liste Select a role die Option Pub/Sub Subscriber aus.

10. Klicken Sie auf die Schaltfläche Save. Der Name des Dienstkontos wird der Dropdown-Liste Pub/Sub Subscriber im Informationsbereich hinzugefügt.

11. Navigieren Sie zur Seite Dashboard in der Cloud Console, und wählen Sie in der Dropdown-Liste Ihr Projekt aus.

12. Klicken Sie auf die Schaltfläche ADD PEOPLE TO THIS PROJECT.

13. Fügen Sie den Namen des Dienstkontos hinzu, den Sie sich notiert haben.

14. Wählen Sie in der Dropdown-Liste Select a role die Option Monitoring Viewer aus.

15. Klicken Sie auf die Schaltfläche Save. Der Name des Dienstkontos wird der Rolle Monitoring Viewer hinzugefügt.

Schritt 3: Stagingbereich erstellen (falls erforderlich)¶

Erstellen Sie mit dem Befehl CREATE STAGE einen externen Stagingbereich, der auf Ihren GCS-Bucket verweist. Snowflake liest Ihre Staging-Datendateien in die Metadaten der externen Tabelle. Alternativ können Sie einen vorhandenen externen Stagingbereich verwenden.

Im folgenden Beispiel wird im aktiven Schema der Benutzersitzung ein Stagingbereich mit dem Namen mystage erstellt. Die Cloudspeicher-URL enthält den Pfad files. Der Stagingbereich verweist auf eine Speicherintegration mit dem Namen my_storage_int.

USE SCHEMA mydb.public;

CREATE STAGE mystage
  URL='gcs://load/files/'
  STORAGE_INTEGRATION = my_storage_int;

Copy

Schritt 4: Pipe mit aktivierter automatischer Erfassung erstellen¶

Erstellen Sie mit dem Befehl CREATE PIPE eine Pipe. Die Pipe definiert die COPY INTO <Tabelle>-Anweisung, mit der Snowpipe Daten aus der Erfassungswarteschlange in die Zieltabelle lädt.

CREATE PIPE <pipe_name>
  AUTO_INGEST = true
  INTEGRATION = '<notification_integration_name>'
  AS
<copy_statement>;

Wobei:

<Pipe-Name>

Bezeichner für die Pipe. Dieser muss für das Schema, in dem die Pipe erstellt wird, eindeutig sein.

Der Bezeichner muss mit einem alphabetischen Zeichen beginnen und darf keine Leer- oder Sonderzeichen enthalten, es sei denn, die gesamte Bezeichnerzeichenfolge wird in doppelte Anführungszeichen gesetzt (z. B. "My object"). Bei Bezeichnern, die in doppelte Anführungszeichen eingeschlossen sind, ist auch die Groß-/Kleinschreibung zu beachten.

INTEGRATION = '<Name_der_Benachrichtigungsintegration>'

Name der Benachrichtigungsintegration an, die zur automatischen Aktualisierung der Metadaten der Verzeichnistabelle unter Verwendung von Google Cloud Pub/Sub-Benachrichtigungen verwendet wird. Eine Benachrichtigungsintegration ist ein Snowflake-Objekt, das als Schnittstelle zwischen Snowflake und Cloud-Meldungswarteschlangendiensten von Drittanbietern dient.

copy_statement

COPY INTO <Tabelle>-Anweisung, mit der Daten aus Warteschlangendateien in eine Snowflake-Tabelle geladen werden. Diese Anweisung dient als Text/Definition für die Pipe und wird in der Ausgabe von SHOW PIPES angezeigt.

Erstellen Sie beispielsweise im Schema snowpipe_db.public eine Pipe, die die Daten aus den im externen (GCS) Stagingbereich mystage bereitgestellten Dateien in die Tabelle mytable lädt:

CREATE PIPE snowpipe_db.public.mypipe
  AUTO_INGEST = true
  INTEGRATION = 'MY_NOTIFICATION_INT'
  AS
COPY INTO snowpipe_db.public.mytable
  FROM @snowpipe_db.public.mystage/path2;

Der INTEGRATION-Parameter verweist auf die Benachrichtigungsintegration my_notification_int, die Sie in Schritt 1: Benachrichtigungsintegration in Snowflake erstellen erstellt haben. Der Integrationsname muss komplett in Großbuchstaben angegeben werden.

Snowpipe mit automatischer Erfassung ist nun konfiguriert!

Wenn dem GCS-Bucket neue Datendateien hinzugefügt werden, weist die Ereignisbenachrichtigung Snowpipe an, diese in die in der Pipe definierte Zieltabelle zu laden.

Schritt 5: Historische Dateien laden¶

Um ein Backlog mit Datendateien zu laden, die vor der Konfiguration von Pub/Sub-Meldungen im externen Stagingbereich vorhanden waren, führen Sie eine ALTER PIPE … REFRESH-Anweisung aus.

Schritt 6: Stagingdateien löschen¶

Löschen Sie die Stagingdateien, nachdem Sie die Daten erfolgreich geladen haben und die Dateien nicht mehr benötigen. Eine Anleitung dazu finden Sie unter Löschen von Stagingdateien, nachdem Snowpipe die Daten geladen hat.