Automatisches Aktualisieren externer Tabellen 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 der Snowflake-Datenbank, in die Ihre Daten geladen werden.

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

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: Externe Tabelle erstellen¶

Erstellen Sie mit dem Befehl CREATE EXTERNAL TABLE eine externe Tabelle.

Erstellen Sie beispielsweise eine externe Tabelle im Schema mydb.public, die JSON-Daten aus Dateien liest, die im Stagingbereich mystage mit dem Pfad path1/ bereitgestellt wurden.

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.

Beachten Sie, dass der Parameter AUTO_REFRESH standardmäßig TRUE lautet:

CREATE OR REPLACE EXTERNAL TABLE ext_table
 INTEGRATION = 'MY_NOTIFICATION_INT'
 WITH LOCATION = @mystage/path1/
 FILE_FORMAT = (TYPE = JSON);

Der externe Stagingbereich mit automatischer Aktualisierung ist jetzt konfiguriert!

Wenn neue oder aktualisierte Datendateien zum GCS-Bucket hinzugefügt werden, weist die Ereignisbenachrichtigung Snowflake an, sie in die Metadaten der externen Tabelle zu scannen.

Schritt 5: Manuelles Aktualisieren der Metadaten der externen Tabelle¶

Aktualisieren Sie die Metadaten der externen Tabelle einmal manuell, indem Sie ALTER EXTERNAL TABLE mit dem Parameter REFRESH verwenden, z. B.:

ALTER EXTERNAL TABLE ext_table REFRESH;

+---------------------------------------------+----------------+-------------------------------+
| file                                        | status         | description                   |
|---------------------------------------------+----------------+-------------------------------|
| files/path1/file1.json                      | REGISTERED_NEW | File registered successfully. |
| files/path1/file2.json                      | REGISTERED_NEW | File registered successfully. |
| files/path1/file3.json                      | REGISTERED_NEW | File registered successfully. |
+---------------------------------------------+----------------+-------------------------------+

Copy

Mit diesem Schritt werden die Metadaten mit der Liste der Dateien im Stagingbereich und Pfad in der externen Tabellendefinition synchronisiert. Außerdem wird durch diesen Schritt sichergestellt, dass die externe Tabelle die Datendateien im angegebenen Stagingbereich und Pfad lesen kann und dass in der Definition der externen Tabelle keine Dateien übersehen wurden.

Wenn die Liste der Dateien in der Spalte file nicht Ihren Erwartungen entspricht, überprüfen Sie die Pfade in der Definition der externen Tabelle und der Definition des externen Stagingbereichs. Jeder Pfad in der externen Tabellendefinition wird an jeden in der Stagingbereichsdefinition angegebenen Pfad angehängt. Weitere Informationen dazu finden Sie unter CREATE EXTERNAL TABLE.

Dieser Schritt stellt sicher, dass die Metadaten mit allen Änderungen an der Dateiliste synchronisiert werden, die seit Schritt 4 vorgenommen wurden. Danach lösen Pub/Sub-Benachrichtigungen automatisch die Aktualisierung der Metadaten aus.

Schritt 6: Konfigurieren der Sicherheit¶

Erteilen Sie mit GRANT <Berechtigungen> für jede zusätzliche Rolle, die zum Abfragen der externen Tabelle verwendet wird, ausreichende Zugriffssteuerungsrechte für die verschiedenen Objekte (d. h. Datenbanken, Schemas, Stagingbereich und Tabelle):