Replikation von Stagingbereichen, Pipes und des Ladeverlaufs¶

Anforderungen¶

Bevor Sie Datenpipeline-Objekte replizieren können, müssen Sie die ETL-Replikation aktivieren. Die Unterstützung der ETL-Replikation ist Bestandteil von BCR-Bundle 2024_01. Sie können BCR-Bundle 2024_01 aktivieren, indem Sie die folgende Anweisung ausführen:

SELECT SYSTEM$ENABLE_BEHAVIOR_CHANGE_BUNDLE('2024_01');

Um beliebige externe Stagingbereiche replizieren zu können, die eine Speicherintegration verwenden, müssen Sie Ihre Replikations-/Failover-Gruppe auch für die Replikation von STORAGE INTEGRATIONS konfigurieren. Andernfalls werden die externen Stagingbereiche ohne die zugeordnete Speicherintegration repliziert.

Sie können eine ALTER REPLICATION GROUP- oder ALTER FAILOVER GROUP-Anweisung verwenden, um diese Eigenschaften für eine bestehende Gruppe zu ändern.

Beispiel:

ALTER FAILOVER GROUP my_failover_group SET
  OBJECT_TYPES = ROLES, INTEGRATIONS
  ALLOWED_INTEGRATION_TYPES = API INTEGRATIONS, STORAGE INTEGRATIONS;

Replikation und Stagingbereiche¶

In diesem Abschnitt wird der aktuelle Stand der Replikationsfunktionalität beschrieben, die Snowflake für verschiedene Typen von Stagingbereichen unterstützt.

Replikation und Pipes¶

In diesem Abschnitt wird beschrieben, welche Replikationsfunktionen derzeit für die verschiedenen Typen von Pipes unterstützt werden.

Snowflake unterstützt die Replikation von folgenden Elementen:

• Pipe-Objekte, einschließlich Pipes für die automatische Datenerfassung und für REST-Endpunkte, die Daten von externen Stagingbereichen laden

• Parameter auf Pipe-Ebene

• Berechtigungen für Pipe-Objekte

Beispiel 1: Replizieren eines benannten internen Stagingbereichs¶

In diesem Beispiel wird gezeigt, wie die Replikation für interne Stagingbereiche funktioniert. Das Beispiel zeigt insbesondere, wie die Verzeichnistabelle die einzige Quelle der Wahrheit für Tabellenmetadaten vor und nach der Replikation ist.

Im ersten Teil des Beispiels werden die folgenden Aufgaben in einem Quellkonto erledigt.

1. Erstellen Sie einen internen Stagingbereich mit dem Namen my_int_stage mit einer aktivierten Verzeichnistabelle, um die im Stagingbereich befindlichen Dateien zu replizieren. Kopieren Sie dann Daten aus einer Tabelle namens my_table in Dateien im Stagingbereich.

   Bemerkung

   Im Beispiel wird die Verzeichnistabelle nach dem Laden von file1 und file2 in den Stagingbereich aktualisiert, um die Tabellenmetadaten mit dem neuesten Satz von Dateien in der Stagingbereichsdefinition für die Verzeichnistabellen zu synchronisieren. Nach dem Laden von file3 findet jedoch keine Aktualisierungsoperation statt.

   CREATE OR REPLACE STAGE my_stage
     DIRECTORY = (ENABLE = TRUE);
   
   COPY INTO @my_stage/folder1/file1 from my_table;
   COPY INTO @my_stage/folder2/file2 from my_table;
   ALTER STAGE my_stage REFRESH;
   
   COPY INTO @my_stage/folder3/file3 from my_table;
   

   Copy

2. Failover-Gruppe erstellen:

   CREATE FAILOVER GROUP my_stage_failover_group
     OBJECT_TYPES = DATABASES
     ALLOWED_DATABASES = my_database_1
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

Im zweiten Teil des Beispiels wird der Replikations- und Failover-Prozess in einem Zielkonto vervollständigt:

1. Erstellen Sie eine Failover-Gruppe als Replikat der Failover-Gruppe im Quellkonto, aktualisieren Sie die Objekte in der neuen Failover-Gruppe, und stufen Sie das Zielkonto zum Quellkonto herauf.

   CREATE FAILOVER GROUP my_stage_failover_group
     AS REPLICA OF myorg.my_account_1.my_stage_failover_group;
   
   ALTER FAILOVER GROUP my_stage_failover_group REFRESH;
   
   ALTER FAILOVER GROUP my_stage_failover_group PRIMARY;
   

   Copy

2. Als Nächstes aktualisieren Sie die Verzeichnistabelle des replizierten Stagingbereichs und kopieren alle Dateien, die von der Verzeichnistabelle in my_stage verfolgt werden, in eine Tabelle namens my_table.

   Bemerkung

   Mit der COPY INTO-Anweisung werden file1 und file2 in die Tabelle geladen, aber nicht file3. Das liegt daran, dass die Verzeichnistabelle nach dem Hinzufügen von file3 im Quellkonto nicht aktualisiert wurde.

   ALTER STAGE my_stage REFRESH;
   
   COPY INTO my_table FROM @my_stage;
   

   Copy

Beispiel 2: Replizieren eines externen Stagingbereichs und einer Speicherintegration¶

Dieses Beispiel zeigt einen Beispiel-Workflow für das Replizieren eines externen Stagingbereichs und der Speicherintegration in ein Zielkonto.

Das Beispiel geht davon aus, dass Sie die folgenden Schritte bereits durchgeführt haben: Konfigurieren des sicheren Zugriffs auf Ihren Amazon S3-Bucket.

Im ersten Teil des Beispiels werden die folgenden Aufgaben in einem Quellkonto erledigt.

1. Erstellen Sie eine Speicherintegration für einen Amazon S3-Bucket in Datenbank my_database_2.

   CREATE STORAGE INTEGRATION my_storage_int
     TYPE = external_stage
     STORAGE_PROVIDER = 's3'
     STORAGE_ALLOWED_LOCATIONS = ('s3://mybucket/path')
     STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/blockedpath')
     ENABLED = true;
   

   Copy

2. Erstellen Sie einen externen Stagingbereich in der Datenbank my_database_2 unter Verwendung der Speicherintegration my_storage_int.

   CREATE STAGE my_ext_stage
     URL = 's3://mybucket/path'
     STORAGE_INTEGRATION = my_storage_int
   

   Copy

3. Erstellen Sie eine Failover-Gruppe, und schließen Sie dabei Datenbank my_database_2 und Speicherintegrationsobjekte ein.

   CREATE FAILOVER GROUP my_external_stage_fg
     OBJECT_TYPES = databases, integrations
     ALLOWED_INTEGRATION_TYPES = storage integrations
     ALLOWED_DATABASES = my_database_2
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

Im zweiten Teil des Beispiels wird der Replikations- und Failover-Prozess in einem Zielkonto vervollständigt:

1. Erstellen Sie eine Failover-Gruppe als Replikat der Failover-Gruppe im Quellkonto, und führen Sie die Aktualisierungsoperation (Refresh) aus:

   CREATE FAILOVER GROUP my_external_stage_fg
     AS REPLICA OF myorg.my_account_1.my_external_stage_fg;
   
   ALTER FAILOVER GROUP my_external_stage_fg REFRESH;
   

   Copy

2. Nachdem Sie die Speicherintegration in das Zielkonto repliziert haben, müssen Sie weitere Schritte ausführen, um die Berechtigungen Ihres Cloudanbieters zu aktualisieren, damit die Replikationsintegration Zugriff auf Ihren Cloudspeicher erhält. Weitere Informationen dazu finden Sie unter Cloudspeicherzugriff für sekundäre Speicherintegrationen konfigurieren.

Beispiel 3: Replizieren einer Auto-Erfassungs-Pipe¶

Dieses Beispiel bietet einen Beispielworkflow für die Replikation einer Pipe, die ein Amazon Simple Notification Service (SNS)-Thema mit Amazon Simple Queue Service (SQS) zum Automatisieren von Snowpipe verwendet.

Im Beispiel wird davon ausgegangen, dass Sie die folgenden Aufgaben bereits erledigt haben:

• Erstellen und Konfigurieren einer Speicherintegration für Amazon S3. Im Beispiel verwenden wir eine Speicherintegration namens my_s3_storage_int.

• Erstellen eines Amazon SNS-Themas und Abonnieren einer Snowflake-SQS-Warteschlange zu Ihrem SNS-Thema.

• Erstellen eines externen Stagingbereichs, der Ihre Speicherintegration referenziert. Im Beispiel verwenden wir einen Stagingbereich namens my_s3_stage. Eine Anleitung dazu finden Sie unter CREATE STAGE.

Beginnen Sie mit den folgenden Aufgaben in einem Quellkonto.

1. Verwenden Sie den Befehl CREATE PIPE, um eine Pipe mit aktivierter automatischer Erfassung zu erstellen, die Daten aus dem externen Stagingbereich in eine Tabelle mit dem Namen mytable lädt.

   CREATE PIPE snowpipe_db.public.mypipe AUTO_INGEST=TRUE
    AWS_SNS_TOPIC='<topic_arn>'
    AS
      COPY INTO snowpipe_db.public.mytable
      FROM @snowpipe_db.public.my_s3_stage
      FILE_FORMAT = (TYPE = 'JSON');
   

   Copy

2. Aktualisieren Sie die Pipe mit einer ALTER PIPE-Anweisung, um Daten der letzten 7 Tage aus dem Stagingbereich zu laden.

   ALTER PIPE mypipe REFRESH;
   

   Copy

3. Verwenden Sie schließlich CREATE FAILOVER GROUP, um eine Failover-Gruppe zu erstellen, die die Replikation von Speicherintegrationen ermöglicht.

   CREATE FAILOVER GROUP my_pipe_failover_group
     OBJECT_TYPES = DATABASES, INTEGRATIONS
     ALLOWED_INTEGRATION_TYPES = STORAGE INTEGRATIONS
     ALLOWED_DATABASES = snowpipe_db
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

Im zweiten Teil des Beispiels wird der Replikations- und Failover-Prozess in einem Zielkonto vervollständigt:

1. Erstellen Sie eine Failover-Gruppe als Replikat der Failover-Gruppe im Quellkonto:

   CREATE FAILOVER GROUP my_pipe_failover_group
     AS REPLICA OF myorg.my_account_1.my_pipe_failover_group;
   

   Copy

2. Führen Sie eine DESCRIBE INTEGRATION-Anweisung aus, um den ARN des AWS-IAM-Benutzers für Ihr Snowflake-Konto in der sekundären Bereitstellung abzurufen.

   Verwenden Sie den ARN, um dem IAM-Benutzer Zugriffsrechte auf Ihren S3 Bucket zu gewähren. Siehe Schritt 5: IAM-Benutzerberechtigungen für den Zugriff auf Bucket-Objekte erteilen.

   DESC INTEGRATION my_s3_storage_int;
   

   Copy

3. Rufen Sie die Systemfunktion SYSTEM$GET_AWS_SNS_IAM_POLICY auf, um eine IAM-Richtlinie zu erstellen, die der neuen SQS-Warteschlange die Berechtigung erteilt, Ihr SNS-Thema zu abonnieren. Snowflake erstellt die neue SQS-Warteschlange in Ihrem Zielkonto, wenn Sie die Failover-Gruppe von Ihrem Quellkonto replizieren.

   SELECT SYSTEM$GET_AWS_SNS_IAM_POLICY('<topic_arn>');
   

   Copy

   topic_arn ist der Amazon Ressource Name (ARN) des SNS-Themas, das Sie in Ihrem Quellkonto für die ursprüngliche Pipe erstellt haben.

   Anschließend müssen Sie die neue Amazon SQS-Warteschlange für Ihr SNS-Thema abonnieren.

4. Aktualisieren Sie die Objekte in Ihrer neuen Failover-Gruppe.

   ALTER FAILOVER GROUP my_pipe_failover_group REFRESH;
   

   Copy

5. Abschließend wird das Zielkonto mit dem Befehl ALTER FAILOVER GROUP zum Quellkonto heraufgestuft.

   ALTER FAILOVER GROUP my_pipe_failover_group PRIMARY;
   

   Copy

   Die mypipe-Pipe beginnt damit, alle Daten zu laden, die seit der letzten Aktualisierung der Failover-Gruppe im Quellkonto bereitgestellt wurden.

   Um sicherzustellen, dass die replizierte Pipe funktioniert, führen Sie mit der COPY-Anweisung der Pipe eine Abfrage auf der Tabelle aus.

   SELECT * FROM mytable;
   

   Copy