Automatisieren von Snowpipe für Amazon S3¶

Schritt 1: Zugriffsberechtigungen für den S3-Bucket konfigurieren¶

Schritt 2: IAM-Rolle in AWS erstellen¶

Um die Zugriffsberechtigungen für Snowflake in der AWS-Managementkonsole zu konfigurieren, gehen Sie wie folgt vor:

1. Melden Sie sich bei der AWS Management Console an.

2. Wählen Sie im Startseiten-Dashboard die Option Identity & Access Management (IAM) aus:

   

3. Wählen Sie im linken Navigationsbereich Roles aus.

4. Wählen Sie Create role aus.

   

5. Wählen Sie Another AWS account als vertrauenswürdigen Entitätstyp aus.

6. Geben Sie im Feld Account ID vorläufig Ihre eigene AWS-Konto-ID ein. Später werden Sie die Vertrauensstellung ändern und Snowflake Zugriff gewähren.

7. Wählen Sie die Option Require external ID aus. Eine externe ID wird verwendet, um einer dritten Partei wie Snowflake Zugriff auf Ihre AWS-Ressourcen (wie S3-Buckets) zu gewähren.

   Geben Sie eine Platzhalter-ID ein, z. B. 0000. In einem späteren Schritt werden Sie die Vertrauensstellung für Ihre IAM-Rolle ändern und die externe ID Ihrer Speicherintegration angeben.

8. Wählen Sie Next aus.

9. Wählen Sie die Richtlinie aus, die Sie in Schritt 1: Zugriffsberechtigungen für den S3-Bucket konfigurieren (unter diesem Thema) erstellt haben.

10. Wählen Sie Next aus.

    

11. Geben Sie einen Namen und eine Beschreibung für die Rolle ein, und wählen Sie Create role aus.

    Sie haben nun eine IAM-Richtlinie für einen Bucket erstellt, eine IAM-Rolle erstellt und die Richtlinie an die Rolle angehängt.

12. Notieren Sie den Wert von Role ARN, der auf der Übersichtsseite der Rolle angegeben ist. Im nächsten Schritt erstellen Sie eine Snowflake-Integration, die auf diese Rolle verweist.

    

Schritt 3: Cloud Storage-Integration in Snowflake erstellen¶

Erstellen Sie mit dem Befehl CREATE STORAGE INTEGRATION eine Speicherintegration. Eine Speicherintegration ist ein Snowflake-Objekt, in dem ein für Ihren S3-Cloudspeicher generierter IAM-Benutzer (Identitäts- und Zugriffsverwaltung) zusammen mit einem optionalen Satz zulässiger oder blockierter Speicherorte (d. h. Buckets) gespeichert wird. Cloudanbieter-Administratoren in Ihrer Organisation erteilen dem generierten Benutzer Berechtigungen für die Speicherorte. Dank dieser Option müssen Benutzer beim Erstellen von Stagingbereichen oder beim Laden von Daten keine Anmeldeinformationen eingeben.

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

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

Wobei:

• integration_name ist der Name der neuen Integration.

• iam_role ist der Amazon Resource Name (ARN) der Rolle, die Sie in Schritt 2: IAM-Rolle in AWS erstellen (unter diesem Thema) erstellt haben.

• bucket ist der Name eines S3-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.

Das folgende Beispiel erstellt eine Integration, die den Zugriff auf alle Buckets im Konto erlaubt, aber den Zugriff auf die definierten sensitivedata-Ordner blockiert.

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

CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::001234567890:role/myrole'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket1/mypath1/sensitivedata/', 's3://mybucket2/mypath2/sensitivedata/');

Copy

Schritt 4: AWS-IAM-Benutzer für Ihr Snowflake-Konto abrufen¶

1. Um den ARN des IAM-Benutzers abzurufen, der für Ihr Snowflake-Konto automatisch erstellt wurde, verwenden Sie den Befehl DESCRIBE INTEGRATION.

   DESC INTEGRATION <integration_name>;
   

   Copy

   Wobei:

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

   Beispiel:

   DESC INTEGRATION s3_int;
   
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
   | property                  | property_type | property_value                                                                 | property_default |
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------|
   | ENABLED                   | Boolean       | true                                                                           | false            |
   | STORAGE_ALLOWED_LOCATIONS | List          | s3://mybucket1/mypath1/,s3://mybucket2/mypath2/                                | []               |
   | STORAGE_BLOCKED_LOCATIONS | List          | s3://mybucket1/mypath1/sensitivedata/,s3://mybucket2/mypath2/sensitivedata/    | []               |
   | STORAGE_AWS_IAM_USER_ARN  | String        | arn:aws:iam::123456789001:user/abc1-b-self1234                                 |                  |
   | STORAGE_AWS_ROLE_ARN      | String        | arn:aws:iam::001234567890:role/myrole                                          |                  |
   | STORAGE_AWS_EXTERNAL_ID   | String        | MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=                               |                  |
   +---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
   

   Copy

2. Notieren Sie sich die Werte der folgenden Eigenschaften:

   Eigenschaft	Beschreibung
   STORAGE_AWS_IAM_USER_ARN	Der AWS-IAM-Benutzer, der für Ihr Snowflake-Konto erstellt wurde, in diesem Beispiel arn:aws:iam::123456789001:user/abc1-b-self1234. Snowflake stellt genau einen IAM-Benutzer für Ihr gesamtes Snowflake-Konto bereit. Alle S3-Speicherintegrationen in Ihrem Konto verwenden diesen IAM-Benutzer.
   STORAGE_AWS_EXTERNAL_ID	Die externe ID, die Snowflake verwendet, um eine Vertrauensstellung mit AWS einzurichten. Wenn Sie beim Erstellen der Speicherintegration keine externe ID (STORAGE_AWS_EXTERNAL_ID) angegeben haben, generiert Snowflake eine ID, die Sie verwenden können.

   Diese Werte geben Sie im nächsten Abschnitt an.

Schritt 5: IAM-Benutzerberechtigungen für den Zugriff auf Bucket-Objekte erteilen¶

Die folgende schrittweise Anleitung beschreibt, wie Sie die IAM-Zugriffsberechtigungen für Snowflake über Ihre AWS Management Console so konfigurieren, dass Sie einen S3-Bucket zum Laden und Entladen von Daten verwenden können:

1. Melden Sie sich bei der AWS Management Console an.

2. Wählen Sie Identity & Access Management (IAM) aus.

   

3. Wählen Sie im linken Navigationsbereich Roles aus.

4. Wählen Sie die Rolle aus, die Sie in Schritt 2: IAM-Rolle in AWS erstellen (unter diesem Thema) erstellt haben.

5. Wählen Sie die Registerkarte Trust relationships aus.

6. Wählen Sie Edit trust relationship aus.

7. Ändern Sie das Richtliniendokument mit den DESC STORAGE INTEGRATION-Ausgabewerten, die Sie in Schritt 4: AWS-IAM-Benutzer für Ihr Snowflake-Konto abrufen (unter diesem Thema) ermittelt haben.

   Richtliniendokument für die IAM-Rolle

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "",
         "Effect": "Allow",
         "Principal": {
           "AWS": "<snowflake_user_arn>"
         },
         "Action": "sts:AssumeRole",
         "Condition": {
           "StringEquals": {
             "sts:ExternalId": "<snowflake_external_id>"
           }
         }
       }
     ]
   }
   

   Copy

   Wobei:

   • snowflake_user_arn ist der von Ihnen notierte STORAGE_AWS_IAM_USER_ARN-Wert.

   • snowflake_external_id ist der von Ihnen notierte STORAGE_AWS_EXTERNAL_ID-Wert.

     In diesem Beispiel hat snowflake_external_id den Wert MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=.

     Bemerkung

     Wenn Sie aus Sicherheitsgründen eine neue Speicherintegration erstellen (oder eine bestehende Speicherintegration mit der CREATE OR REPLACE STORAGE INTEGRATION-Syntax neu erstellen) und dabei keine externe ID angeben, hat die neue Speicherintegration eine andere externe ID und kann daher die Vertrauensstellung nicht auflösen, es sei denn, Sie aktualisieren die Vertrauensrichtlinie.

8. Wählen Sie die Schaltfläche Update Trust Policy aus. Die Änderungen werden gespeichert.

Schritt 1: Stagingbereich erstellen (falls erforderlich)¶

Erstellen Sie mit dem Befehl CREATE STAGE einen externen Stagingbereich, der auf Ihren S3-Bucket verweist. Snowpipe ruft Ihre Datendateien aus dem Stagingbereich ab und stellt sie vorübergehend in die Warteschlange, bevor sie in die Zieltabelle geladen werden. 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 snowpipe_db.public;

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

Copy

Schritt 2: 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.

Im folgenden Beispiel wird im aktiven Schema der Benutzersitzung eine Pipe mit dem Namen mypipe erstellt. Die Pipe lädt die Daten aus den im Stagingbereich mystage bereitgestellten Dateien in die Tabelle mytable:

create pipe snowpipe_db.public.mypipe auto_ingest=true as
  copy into snowpipe_db.public.mytable
  from @snowpipe_db.public.mystage
  file_format = (type = 'JSON');

Copy

Der Parameter AUTO_INGEST=true gibt an, dass die von einem S3-Bucket an eine SQS-Warteschlange gesendeten Ereignisbenachrichtigungen gelesen werden, sobald neue Daten ladebereit sind.

Schritt 3: Sicherheit konfigurieren¶

Vergeben Sie für jeden Benutzer, der kontinuierliches Laden von Daten mit Snowpipe durchführt, ausreichende Zugriffssteuerungsrechte an den Objekten für das Laden von Daten (d. h. Zieldatenbank, Schema und Tabelle; Stagingobjekt und Pipe).

Die Verwendung von Snowpipe erfordert eine Rolle mit den folgenden Berechtigungen:

Verwenden Sie den Befehl GRANT <Berechtigungen>, um der Rolle Berechtigungen zu erteilen.

Erstellen Sie beispielsweise eine Rolle mit dem Namen snowpipe_role, die auf einen Satz von snowpipe_db.public-Datenbankobjekten sowie auf eine Pipe mit dem Namen mypipe zugreifen kann. Weisen Sie die Rolle anschließend einem Benutzer zu:

-- Create a role to contain the Snowpipe privileges
use role securityadmin;

create or replace role snowpipe_role;

-- Grant the required privileges on the database objects
grant usage on database snowpipe_db to role snowpipe_role;

grant usage on schema snowpipe_db.public to role snowpipe_role;

grant insert, select on snowpipe_db.public.mytable to role snowpipe_role;

grant usage on stage snowpipe_db.public.mystage to role snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = true;

-- Grant the OWNERSHIP privilege on the pipe object
grant ownership on pipe snowpipe_db.public.mypipe to role snowpipe_role;

-- Grant the role to a user
grant role snowpipe_role to user jsmith;

-- Set the role as the default role for the user
alter user jsmith set default_role = snowpipe_role;

-- Resume the pipe
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = false;

Copy

Schritt 4: Ereignisbenachrichtigungen konfigurieren¶

Konfigurieren Sie Ereignisbenachrichtigungen für Ihren S3-Bucket, um Snowpipe zu benachrichtigen, sobald neue Daten zum Laden verfügbar sind. Das Feature zur automatischen Erfassung stützt sich auf SQS-Warteschlangen, um Ereignisbenachrichtigungen von S3 an Snowpipe zu übermitteln.

Zur Vereinfachung der Verwendung werden Snowpipe SQS-Warteschlangen von Snowflake angelegt und verwaltet. Die Ausgabe des Befehls SHOW PIPES zeigt den Amazon Resource Name (ARN) Ihrer SQS-Warteschlange an.

1. Führen Sie den Befehl SHOW PIPES aus:

   SHOW PIPES;
   

   Copy

   Beachten Sie den ARN der SQS-Warteschlange für den Stagingbereich in der Spalte notification_channel. Kopieren Sie den ARN an einen geeigneten Speicherort.

   Bemerkung

   Gemäß AWS-Richtlinien weist Snowflake genau eine SQS-Warteschlange pro S3-Bucket zu. Diese SQS-Warteschlange kann von mehreren Buckets in demselben AWS-Konto gemeinsam genutzt werden. Die SQS-Warteschlange koordiniert Benachrichtigungen für alle Pipes, die die externen Stagingbereiche für den S3-Bucket mit den Zieltabellen verbinden. Wenn eine Datendatei in den Bucket hochgeladen wird, führen alle Pipes, die mit dem Verzeichnispfad des Stagingbereichs übereinstimmen, ein einmaliges Laden der Datei in die entsprechenden Zieltabellen durch.

2. Melden Sie sich bei der Amazon S3-Konsole an.

3. Konfigurieren Sie anhand der Anweisungen in der Amazon S3-Dokumentation eine Ereignisbenachrichtigung für Ihren S3-Bucket. Füllen Sie die Felder wie folgt aus:

   • Name: Name der Ereignisbenachrichtigung (z. B. Auto-ingest Snowflake).

   • Events: Wählen Sie die Option ObjectCreate (All) aus.

   • Send to: Wählen Sie SQS Queue in der Dropdown-Liste aus.

   • SQS: Wählen Sie Add SQS queue ARN in der Dropdown-Liste aus.

   • SQS queue ARN: Fügen Sie den SQS-Warteschlangennamen aus der SHOW PIPES-Ausgabe ein.

Snowpipe mit automatischer Erfassung ist nun konfiguriert!

Wenn dem S3-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¶

Informationen zum Laden aller Backlogs mit Datendateien, die vor der Konfiguration von SQS-Benachrichtigungen im externen Stagingbereich vorhanden waren, finden Sie unter Laden historischer Daten.

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.

Vorbereitung: Amazon-SNS-Thema und Abonnement erstellen.¶

1. Erstellen Sie ein SNS-Thema in Ihrem AWS-Konto, um alle Nachrichten für den Snowflake-Stagingbereich in Ihrem S3-Bucket zu verwalten.

2. Abonnieren Sie Ihre Zieldestinationen für die S3-Ereignisbenachrichtigungen (z. B. andere SQS-Warteschlangen oder AWS Lambda-Workloads) zu diesem Thema. SNS veröffentlicht Ereignisbenachrichtigungen für Ihren Bucket an alle Abonnenten des Themas.

Anweisungen dazu finden Sie in der SNS-Dokumentation.

Schritt 1: Snowflake-SQS-Warteschlange zum SNS-Thema abonnieren¶

1. Anmelden bei der AWS-Managementkonsole.

2. Wählen Sie auf dem Startseiten-Dashboard die Option Simple Notification Service (SNS) aus.

3. Wählen Sie im linken Navigationsbereich Topics aus.

4. Suchen Sie das Thema für Ihren S3-Bucket. Beachten Sie das Thema ARN.

5. Fragen Sie mithilfe eines Snowflake-Clients die SYSTEM$GET_AWS_SNS_IAM_POLICY-Systemfunktion mit Ihrem SNS-Thema ARN ab:

   select system$get_aws_sns_iam_policy('<sns_topic_arn>');
   

   Copy

   Die Funktion gibt eine IAM-Richtlinie zurück, die einer Snowflake-SQS-Warteschlange die Berechtigung zum Abonnieren des SNS-Themas erteilt.

   Beispiel:

   select system$get_aws_sns_iam_policy('arn:aws:sns:us-west-2:001234567890:s3_mybucket');
   
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | SYSTEM$GET_AWS_SNS_IAM_POLICY('ARN:AWS:SNS:US-WEST-2:001234567890:S3_MYBUCKET')                                                                                                                                                                   |
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   | {"Version":"2012-10-17","Statement":[{"Sid":"1","Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"},"Action":["sns:Subscribe"],"Resource":["arn:aws:sns:us-west-2:001234567890:s3_mybucket"]}]}                 |
   +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
   

   Copy

6. Kehren Sie zur AWS Management Console zurück. Wählen Sie im linken Navigationsbereich Topics aus.

7. Wählen Sie das Thema für Ihren S3-Bucket aus, und klicken Sie auf die Schaltfläche Edit. Die Seite Edit wird angezeigt.

8. Klicken Sie auf Access policy - Optional, um diesen Bereich der Seite zu erweitern.

9. Fügen Sie den IAM-Richtlinienzusatz aus den SYSTEM$GET_AWS_SNS_IAM_POLICY-Funktionsergebnissen in das JSON-Dokument ein.

   Beispiel:

   Ursprüngliche IAM-Richtlinie (abgekürzt):

   {
     "Version":"2008-10-17",
     "Id":"__default_policy_ID",
     "Statement":[
        {
           "Sid":"__default_statement_ID",
           "Effect":"Allow",
           "Principal":{
              "AWS":"*"
           }
           ..
        }
      ]
    }
   

   Copy

   Zusammengeführte IAM-Richtlinie:

   {
     "Version":"2008-10-17",
     "Id":"__default_policy_ID",
     "Statement":[
        {
           "Sid":"__default_statement_ID",
           "Effect":"Allow",
           "Principal":{
              "AWS":"*"
           }
           ..
        },
        {
           "Sid":"1",
           "Effect":"Allow",
           "Principal":{
             "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
            },
            "Action":[
              "sns:Subscribe"
            ],
            "Resource":[
              "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
            ]
        }
      ]
    }
   

   Copy

10. Fügen Sie eine zusätzliche Richtlinienberechtigung hinzu, damit S3 Ereignisbenachrichtigungen für den Bucket im SNS-Thema veröffentlichen kann.

    Beispiel (unter Verwendung des in diesen Anweisungen verwendeten SNS-Thema-ARN und S3-Bucket):

    {
        "Sid":"s3-event-notifier",
        "Effect":"Allow",
        "Principal":{
           "Service":"s3.amazonaws.com"
        },
        "Action":"SNS:Publish",
        "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
        "Condition":{
           "ArnLike":{
              "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
           }
        }
     }
    

    Copy

    Zusammengeführte IAM-Richtlinie:

    {
      "Version":"2008-10-17",
      "Id":"__default_policy_ID",
      "Statement":[
         {
            "Sid":"__default_statement_ID",
            "Effect":"Allow",
            "Principal":{
               "AWS":"*"
            }
            ..
         },
         {
            "Sid":"1",
            "Effect":"Allow",
            "Principal":{
              "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
             },
             "Action":[
               "sns:Subscribe"
             ],
             "Resource":[
               "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
             ]
         },
         {
            "Sid":"s3-event-notifier",
            "Effect":"Allow",
            "Principal":{
               "Service":"s3.amazonaws.com"
            },
            "Action":"SNS:Publish",
            "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
            "Condition":{
               "ArnLike":{
                  "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
               }
            }
          }
       ]
     }
    

    Copy

11. Klicken Sie auf die Schaltfläche Save changes.

Schritt 2: Stagingbereich erstellen (falls erforderlich)¶

Erstellen Sie mit dem Befehl CREATE STAGE einen externen Stagingbereich, der auf Ihren S3-Bucket verweist. Snowpipe ruft Ihre Datendateien aus dem Stagingbereich ab und stellt sie vorübergehend in die Warteschlange, bevor sie in die Zieltabelle geladen werden.

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:

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

Copy

Schritt 3: 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. Identifizieren Sie in der COPY-Anweisung den SNS-Thema-ARN aus Voraussetzung: Amazon-SNS-Thema und Abonnement erstellen.

Im folgenden Beispiel wird im aktiven Schema der Benutzersitzung eine Pipe mit dem Namen mypipe erstellt. Die Pipe lädt die Daten aus den im Stagingbereich mystage bereitgestellten Dateien in die Tabelle mytable:

create pipe snowpipe_db.public.mypipe
  auto_ingest=true
  aws_sns_topic='<sns_topic_arn>'
  as
    copy into snowpipe_db.public.mytable
    from @snowpipe_db.public.mystage
  file_format = (type = 'JSON');

Copy

Wobei:

AUTO_INGEST = true

Gibt an, dass die von einem S3-Bucket an eine SQS-Warteschlange gesendeten Ereignisbenachrichtigungen gelesen werden, sobald neue Daten ladebereit sind.

AWS_SNS_TOPIC = '<sns_topic_arn>'

Gibt den ARN für das SNS-Thema Ihres S3-Buckets an, z. B. arn:aws:sns:us-west-2:001234567890:s3_mybucket im aktuellen Beispiel. Mit der CREATE PIPE-Anweisung wird die Snowflake-SQS-Warteschlange für das angegebene SNS-Thema abonniert. Beachten Sie, dass die Pipe nur Dateien in die Erfassungswarteschlange kopiert, die durch Ereignisbenachrichtigungen über das SNS-Thema ausgelöst wurden.

Um einen der Parameter aus einer Pipe zu entfernen, muss die Pipe derzeit mit der Syntax CREATE OR REPLACE PIPE neu erstellt werden.

Schritt 4: Sicherheit konfigurieren¶

Vergeben Sie für jeden Benutzer, der kontinuierliches Laden von Daten mit Snowpipe durchführt, ausreichende Zugriffssteuerungsrechte an den Objekten für das Laden von Daten (d. h. Zieldatenbank, Schema und Tabelle; Stagingobjekt und Pipe).

Die Verwendung von Snowpipe erfordert eine Rolle mit den folgenden Berechtigungen:

Verwenden Sie den Befehl GRANT <Berechtigungen>, um der Rolle Berechtigungen zu erteilen.

Erstellen Sie beispielsweise eine Rolle mit dem Namen snowpipe_role, die auf einen Satz von snowpipe_db.public-Datenbankobjekten sowie auf eine Pipe mit dem Namen mypipe zugreifen kann. Weisen Sie die Rolle anschließend einem Benutzer zu:

-- Create a role to contain the Snowpipe privileges
use role securityadmin;

create or replace role snowpipe_role;

-- Grant the required privileges on the database objects
grant usage on database snowpipe_db to role snowpipe_role;

grant usage on schema snowpipe_db.public to role snowpipe_role;

grant insert, select on snowpipe_db.public.mytable to role snowpipe_role;

grant usage, read on stage snowpipe_db.public.mystage to role snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = true;

-- Grant the OWNERSHIP privilege on the pipe object
grant ownership on pipe snowpipe_db.public.mypipe to role snowpipe_role;

-- Grant the role to a user
grant role snowpipe_role to user jsmith;

-- Set the role as the default role for the user
alter user jsmith set default_role = snowpipe_role;

-- Resume the pipe
alter pipe mypipe SET PIPE_EXECUTION_PAUSED = false;

Copy

Snowpipe mit automatischer Erfassung ist nun konfiguriert!

Wenn dem S3-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¶

Informationen zum Laden aller Backlogs mit Datendateien, die vor der Konfiguration von SQS-Benachrichtigungen im externen Stagingbereich vorhanden waren, finden Sie unter Laden historischer Daten.

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.

Schritt 1: Amazon SNS-Thema erstellen¶

Führen Sie die unter Vorbereitung: Amazon-SNS-Thema und Abonnement erstellen (unter diesem Thema) beschriebenen Schritte aus.

Schritt 2: EventBridge-Regel zum Abonnieren von S3-Buckets und zum Senden von Benachrichtigungen an SNS-Themen erstellen¶

• Aktivieren Sie Amazon EventBridge für S3-Buckets.

• Erstellen Sie EventBridge-Regeln, die das Senden von Benachrichtigungen an das in Schritt 1 erstellte SNS-Thema steuern.

Option 3: Konfigurieren von Amazon SNS zum Automatisieren von Snowpipe mithilfe von SQS-Benachrichtigungen¶

Führen Sie die unter Option 2: Konfigurieren von Amazon SNS zum Automatisieren von Snowpipe mithilfe von SQS-Benachrichtigungen (unter diesem Thema) beschriebenen Schritte aus.