Ausfallsicherheit an mehreren Standorten für Datenpipelines¶

Anforderungen¶

• Edition: Snowflake Business Critical Edition (oder höher).

• Unterstützte Datenaufnahmemethoden: Dieses Feature unterstützt ausschließlich das dateibasierte Laden von Daten über Snowpipe (automatische Datenaufnahme) und COPY INTO <table>. Openflow oder Snowpipe Streaming wird nicht unterstützt.

• Identische Pfadstrukturen: Damit Ihre Pipelines nach dem Failover neue Dateien finden können, müssen Sie diese Dateien mit derselben Hierarchie, Ordnerstruktur und demselben relativen Pfad wie Ihr primärer Speicherort in den sekundären Speicherort schreiben.

Hinweise¶

• Abrechnung: Für dieses Feature fallen Standardreplikationsgebühren (Datentransfer und Computeressourcen) an, die Ihrem Zielkonto in Rechnung gestellt werden.

• Ausfallzeit der Stagingbereichsänderung: Durch die Änderung derRELATIVE_URL-Eigenschaft in einem vorhandenen Stagingbereich werden abhängige Objekte ungültig gemacht und die Erfassung angehalten. Wir empfehlen, während der Einrichtung neue Stagingbereiche zu erstellen, um Ausfallzeiten zu vermeiden.

• Integration von Benachrichtigungen für mehrere Warteschlangen (MQNI): Die Verwendung derselben aktiven Warteschlange sowohl in Quell- als auch in Zielkonten wird nicht unterstützt. Dies kann zu einem Verlust von Benachrichtigungen führen. Snowflake prüft nicht, ob von mehreren Konten dieselbe Warteschlange verwendet wird.

• Verzeichnistabelle: Das Erstellen einer Verzeichnistabelle in einem Stagingbereich mithilfe von MLSI wird derzeit nicht unterstützt.

Replikationsverhalten¶

• Asynchrone Replikation: Snowflake repliziert Ihre Tabellen und den Ladeverlauf Ihrer Pipeline zusammen in exakt demselben Snapshot. Da sie synchronisiert sind, führt ein Ausfall nicht zu duplizierten Daten. Wenn Ihre Sekundärdatenbank vier Stunden im Rückstand ist, liegen auch die Tabellendaten vier Stunden zurück, und die Verarbeitung von vier Stunden an in der Warteschlange befindlichen Benachrichtigungen bringt einfach die Tabelle auf den neuesten Stand.

• Vermeiden von Datenverlusten durch doppeltes Schreiben: Ihr Wiederherstellungspunkt-Objekt (RPO, Recovery Point Objective) wird durch Ihr Aktualisierungsintervall für die Replikation bestimmt. Um Datenverluste während eines Failovers zu vermeiden, muss die Nachrichtenaufbewahrungsfrist Ihrer sekundären Cloud-Meldungswarteschlange länger sein als Ihr Replikationsintervall. Wenn Ihre Warteschlange Meldungen löscht, bevor Ihre geplante Replikation nachgeholt hat, werden diese Dateien beim Failover nicht erfasst.

• ** Risiko für Datenverlust bei einzelnem Schreiben:** Wenn Sie das Routing mit einzelnem Schreiben verwenden, sind alle Dateien, die nach Ihrer letzten erfolgreichen Replikation am primären Speicherort verarbeitet werden, dem sekundären Speicherort vollständig unbekannt. Beim Failover fehlen diese Daten vorübergehend in Ihrem Zielkonto.

1. Doppelte Schreibvorgänge – empfohlen¶

Ihre Ersteller-Anwendung schreibt Dateien gleichzeitig in Ihre primären und sekundären Cloudspeicher-Buckets. Die sekundäre Nachrichtenwarteschlange sammelt Benachrichtigungen, aber Snowflake verarbeitet sie nicht, da die sekundäre Datenbank schreibgeschützt ist.

• Was beim Failover passiert: Die Sekundärdatenbank wird beschreibbar. Snowpipe liest die sekundäre Warteschlange und verwendet den replizierten Ladeverlauf, um Dateien zu deduplizieren. Wenn ein Ausfall verhindert hat, dass eine Datei am primären Speicherort beendet wurde, liest die sekundäre Pipeline die Benachrichtigung aus der sekundären Warteschlange, erkennt, dass die Datei im Ladeverlauf fehlt, und nimmt sie auf.

• Was beim Failback passiert: Wenn der primäre Speicherort wiederhergestellt wird, Sie die Failover-Gruppe aktualisieren und dann ein Failback ausführen, beginnt Snowpipe automatisch mit der Aufnahme neuer Dateien, da der Ladeverlauf während der Failback-Vorbereitung vom sekundären Konto synchronisiert wurde.

• Ergebnis: Keine fehlenden Daten, keine Duplikate. Snowflake führt den Abgleich automatisch in beide Richtungen aus.

• Erforderliche Aktion: Keine, über die Standardsynchronisation der Failback-Replikation hinaus (ALTER FAILOVER GROUP … REFRESH), um sicherzustellen, dass das primäre Konto über den neuesten Ladeverlauf verfügt.

2. Routing mit einzelnem Schreiben¶

Ihr Ersteller schreibt nur in den primären Cloudspeicher. Bei einem Ausfall leiten Sie den Ersteller um, um mit dem Schreiben neuer Dateien in den sekundären Cloudspeicher zu beginnen.

• Was beim Failover passiert: Das sekundäre Konto beginnt sofort mit der Verarbeitung neuer Dateien, die an den sekundären Bucket weitergeleitet werden. Alle Dateien, die am betroffenen Primärspeicherort ausgeführt werden und eingeschlossen sind, bleiben jedoch vorübergehend zurück.

• Was beim Failback passiert: Wenn der primäre Speicherort wiederhergestellt ist und Sie ein Failback für Ihr primäres Snowflake-Konto ausführen, verarbeitet Snowpipe automatisch alle Dateibenachrichtigungen, die vor dem Ausfall erfolgreich die Warteschlange erreicht haben.

• Ergebnis: Keine Duplikate. Alle Dateien, bei denen die Cloudbenachrichtigung aufgrund des Ausfalls jedoch nicht vollständig generiert werden konnte (oder der Ausfall die Richtlinien zur Nachrichtenaufbewahrung Ihrer Warteschlange überschritten hat), erfordern ein manuelles Eingreifen.

• Erforderliche Aktion: Vergleichen Sie nach dem Failback Ihren primären Speicher-Bucket mit der COPY_HISTORY-Ansicht in Snowflake, um fehlende Dateien zu identifizieren. Führen Sie ALTER PIPE … REFRESH oder einen manuellen COPY INTO-Befehl aus, um diese spezifischen übrig gebliebenen Dateien zu laden.

Schritt 1: Speicherintegration mit mehreren Speicherorten (MLSI) erstellen¶

Um eine Speicherintegration mit mehreren Speicherorten zu konfigurieren, folgen Sie den Standardschritten für die Konfiguration einer Speicherintegration mit ein paar Unterschieden, die in diesem Abschnitt beschrieben werden.

Erstellen Sie in Ihrem Quellkonto die MLSI, indem Sie Werte für jeden Speicherort in der STORAGE_LOCATIONS-Liste angeben. Sie können Cloudanbieter für Cloud-übergreifende Einrichtungen mischen.

CREATE STORAGE INTEGRATION my_mlsi
  TYPE = EXTERNAL_STAGE
  STORAGE_LOCATIONS =
  (
    (
      NAME = 'my-s3-us-west-1'
      STORAGE_PROVIDER = 'S3'
      STORAGE_BASE_URL = 's3://myBucketWest'
      STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::12345:role/myrole'
      STORAGE_AWS_EXTERNAL_ID = 'mlsi-external-id'
      ENCRYPTION = ( TYPE = 'AWS_SSE_S3' )
    ),
    (
      NAME = 'my-s3-us-east-1'
      STORAGE_PROVIDER = 'S3'
      STORAGE_BASE_URL = 's3://myBucketEast'
      STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::67890:role/myrole'
      STORAGE_AWS_EXTERNAL_ID = 'mlsi-external-id'
      ENCRYPTION = ( TYPE = 'AWS_SSE_S3' )
    )
  )
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('*')
  ACTIVE = 'my-s3-us-west-1';

Wobei:

• STORAGE_LOCATIONS: Gibt eine Liste von einem oder mehreren Speicherorten (S3-Bucket, GCS-Bucket oder Azure-Container) für die Speicherintegration an. Informationen zum Anzeigen der Parameter für jeden Cloudanbieter finden Sie unter Cloudanbieterparameter (cloudProviderParams) auf der CREATE STORAGE INTEGRATION-Referenzseite.

• NAME: Zeichenfolge, die den Bezeichner (Name) für den Speicherort angibt.

• ENCRYPTION: Gibt die Verschlüsselung für den Speicherort an. Sie müssen die Verschlüsselung für den Speicherort auf der Ebene der Speicherintegration angeben und nicht auf der Ebene des Stagingbereichs. Nur erforderlich zum Laden aus verschlüsselten Dateien; nicht erforderlich, wenn der Speicherort und die Dateien unverschlüsselt sind. Informationen zum Anzeigen der Verschlüsselungsoptionen für jeden Cloudanbieter finden Sie unter ENCRYPTION auf der CREATE STAGE-Referenzseite.

• ACTIVE: Gibt den Namen des Speicherorts an, der als aktiver Speicherort für die Speicherintegration im aktuellen Konto festgelegt werden soll.

  Für den aktiven Speicherort in Ihrem Quellkonto müssen Sie die Zugriffssteuerung einrichten und Snowflake die Berechtigung zum Zugriff auf Ihren Speicher erteilen. Verwenden Sie die Anweisungen in den folgenden Themen:

  • Konfigurieren einer Snowflake-Speicherintegration für Zugriff auf Amazon S3

  • Konfigurieren einer Integration für Google Cloud Storage

  • Konfigurieren eines Azure-Containers zum Laden von Daten

Schritt 2: MLSI mit einem externen Stagingbereich verknüpfen¶

Es wird dringend empfohlen, einen neuen Stagingbereich zu erstellen, anstatt einen bestehenden Stagingbereich zu ändern.

Verwenden Sie den Befehl CREATE STAGE, um die von Ihnen erstellte Speicherintegration mit mehreren Speicherorten mit einem oder mehreren externen Stagingbereichen zu verknüpfen:

CREATE STAGE my_ext_stage
  RELATIVE_URL = '/my_folder/my_sub_folder/'
  STORAGE_INTEGRATION = 'my_mlsi';

Wobei:

• RELATIVE_URL: Der relative Pfad zu Ihrem Speicherort des externen Stagingbereichs von dem in Ihrer Speicherintegration definierten Speicherort. Damit Ihre Pipelines nach dem Failover neue Dateien finden können, müssen Sie diese Dateien mit derselben Hierarchie, Ordnerstruktur und demselben relativen Pfad wie Ihr primärer Speicherort in den sekundären Speicherort schreiben.

• STORAGE_INTEGRATION: Der Name Ihrer Speicherintegration für mehrere Speicherorte.

Beispiel:

ALTER STAGE my_ext_stage SET
  RELATIVE_URL = '/my_folder/my_sub_folder/'
  STORAGE_INTEGRATION = 'my_mlsi';

Schritt 3: Eine Benachrichtigungsintegration für mehrere Warteschlangen konfigurieren (MQNI)¶

Wenn Sie das automatisierte Laden von Daten über Cloudmessaging verwenden und eine Speicherintegration mit mehreren Speicherorten für Ihren externen Stagingbereich konfiguriert haben, müssen Sie auch eine Benachrichtigungsintegration für mehrere Warteschlangen für ein nahtloses Failover Ihrer Snowpipe-Pipelines verwenden.

Für jede Warteschlange, die Sie für die Benachrichtigungsintegration definieren, müssen Sie Ihren Messagingdienst mithilfe der Schritte unter folgenden Themen vorbereiten:

• Konfigurieren von Amazon SNS zum Automatisieren von Snowpipe mithilfe von SQS-Benachrichtigungen. Erstellen Sie ein SNS-Thema für jede AWS-Region, in der sich die Speicherorte Ihrer MLSI befinden.

• Konfigurieren der Automatisierung mit GCS Pub/Sub

• Konfigurieren der Automatisierung mit Azure Event Grid

CREATE NOTIFICATION INTEGRATION my_mqni
  ENABLED = TRUE
  TYPE = MULTI_QUEUE
  DIRECTION = INBOUND
  QUEUES = (
    (
      NAME = 'my-us-west-1'
      NOTIFICATION_PROVIDER = AWS_SNS
      AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-west'
    ),
    (
      NAME = 'my-us-east-1'
      NOTIFICATION_PROVIDER = AWS_SNS
      AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-east'
    )
  )
  ACTIVE = 'my-us-west-1';

CREATE PIPE my_pipe
  AUTO_INGEST = TRUE
  INTEGRATION = my_mqni
  AS COPY INTO my_table FROM @my_ext_stage/my_pipe/;

SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  '<new_mqni_name>',
  '<original_sns_topic_arn_or_int_name>',
  '<new_sns_topic_arn_or_int_name>'
)

SELECT SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  'my_mqni',
  'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-west',
  'arn:aws:sns:us-east-1:67890:my-snowpipe-mlsi-east'
);

SELECT SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  'my_azure_mqni',
  'my_azure_ni_1',
  'my_azure_ni_2'
);

Schritt 4: Ihre MLSI und MQNI auf das Zielkonto replizieren¶

1. To replicate your multi-location storage integration and multi-queue notification integration, alter your existing replication or failover group to include STORAGE INTEGRATIONS and NOTIFICATION INTEGRATIONS in the ALLOWED_INTEGRATION_TYPES list.

Verwenden Sie zum Beispiel den Befehl ALTER FAILOVER GROUP:

ALTER FAILOVER GROUP my_fg SET
  OBJECT_TYPES = DATABASES, ROLES, INTEGRATIONS
  ALLOWED_INTEGRATION_TYPES = API INTEGRATIONS, STORAGE INTEGRATIONS,
    NOTIFICATION INTEGRATIONS;

2. Führen Sie dann in Ihrem Zielkonto eine Aktualisierungsoperation aus:

ALTER FAILOVER GROUP my_fg REFRESH;

Schritt 5: Aktive Status im Zielkonto konfigurieren¶

Um nach der Aktualisierungsoperation ein nahtloses Failover während eines tatsächlichen Ausfalls sicherzustellen, konfigurieren Sie den aktiven Speicherort und die Warteschlange (bei Verwendung einer Benachrichtigungsintegration) in Ihrem Zielkonto.

In Ihrem Zielkonto:

1. Für den Speicherort, den Sie als aktiven Speicherort in Ihrem Zielkonto festlegen möchten, verwenden Sie die Anweisungen unter folgenden Themen, um Snowflake Zugriff auf Ihren Speicher zu gewähren:

   • Option 1: Konfigurieren einer Snowflake-Speicherintegration für Zugriff auf Amazon S3

   • Konfigurieren einer Integration für Google Cloud Storage

   • Konfigurieren eines Azure-Containers zum Laden von Daten

2. Sekundärspeicher aktivieren: Stellen Sie die MLSI ein, um Ihren sekundären Backup-Speicherort im Zielkonto zu verwenden.

   CopyExpand

   ALTER STORAGE INTEGRATION my_mlsi SET ACTIVE = 'my-s3-us-east-1';
   

   Show lessSee more

   Scroll to top

3. Wenn Sie eine Benachrichtigungsintegration für mehrere Warteschlangen verwenden, erteilen Sie Snowflake die Berechtigung, auf Ihren Messagingdienst für die Warteschlange zuzugreifen, die Sie in Ihrem Zielkonto als aktiv einstellen möchten. Folgen Sie den Anweisungen für Ihren Cloudanbieter:

   • Für AWS siehe „Snowflake-SQS-Warteschlange zum SNS-Thema abonnieren“ unter Vorbereitung: Amazon-SNS-Thema und Abonnement erstellen..

   • Für Google Cloud siehe „Snowflake Zugriff auf das Pub/Sub-Abonnement gewähren“ unter Konfigurieren von automatischem Snowpipe mit GCS Pub/Sub.

   • Für Azure siehe „Snowflake Zugriff auf die Speicherwarteschlange gewähren“ unter Konfigurieren der Automatisierung mit Azure Event Grid.

4. Aktivieren Sie die sekundäre Warteschlange (bei Verwendung von MQNI): Setzen Sie die aktive Warteschlange auf Ihren sekundären Speicherort im Zielkonto.

   CopyExpand

   ALTER INTEGRATION my_mqni
     SET ACTIVE = 'MY_MQNI-queue2';
   

   Show lessSee more

   Scroll to top

1. Aktive Integrationsstatus verifizieren¶

Vergewissern Sie sich, dass Ihre Integrationen auf den richtigen Speicher und die richtigen Warteschlangen verweisen, indem Sie deren Eigenschaften überprüfen. Suchen Sie nach der ACTIVE-Eigenschaft in der Ausgabe:

-- Check the active storage location
DESCRIBE STORAGE INTEGRATION my_mlsi;

-- Check the active message queue
DESCRIBE INTEGRATION my_mqni;

2. Pipestatus überprüfen (nur Snowpipe)¶

Verwenden Sie die Funktion SYSTEM$PIPE_STATUS, um sicherzustellen, dass Ihre Pipe ausgeführt wird, und um zu prüfen, ob sie aktiv neue Dateien von Ihrem sekundären Speicherort in die Warteschlange stellt.

SELECT SYSTEM$PIPE_STATUS('my_pipe');

Suchen Sie nach „executionState“:“RUNNING“, und überprüfen Sie „pendingFileCount“, um zu bestätigen, dass es aktiv neue Dateien erkennt, die in Ihrem sekundären Bucket abgelegt werden.

3. Erfolgreiche Datenaufnahme validieren (Ladeverlauf)¶

Um sicherzustellen, dass die Daten ohne Fehler oder Duplikate geladen werden, fragen Sie die COPY_HISTORY-Ansicht ab. Dies zeigt genau, welche Dateien aufgenommen wurden, ihren Quellpfad und wann sie geladen wurden.

SELECT file_name, status, row_count, last_load_time
FROM TABLE(information_schema.copy_history(
  table_name => 'my_table',
  start_time => DATEADD(hours, -1, CURRENT_TIMESTAMP())
));

Überprüfen Sie, ob die file_name-Pfade Ihren aktiven Speicherort angeben und der Status als LOADED angezeigt wird.