Umgekehrtes ETL verwenden, um Daten aus BigQuery in Spanner Graph zu laden

In diesem Dokument wird beschrieben, wie Sie Pipelines für Reverse Extrahieren, Transformieren und Laden (ETL) verwenden, um Graphdaten aus BigQuery nach Spanner Graph zu verschieben und kontinuierlich zu synchronisieren. Dabei werden die folgenden wichtigen Aspekte berücksichtigt:

Informationen zum Exportieren von Daten aus BigQuery nach Spanner mit Reverse-ETL finden Sie unter Daten nach Spanner exportieren.

BigQuery führt komplexe Datenbearbeitung im großen Maßstab als Analyseplattform durch, während Spanner für Anwendungsfälle optimiert ist, die einen hohen QPS und eine niedrige Serving-Latenz erfordern. Spanner Graph und BigQuery lassen sich effektiv integrieren, um Grafdaten in BigQuery-Analysepipelines vorzubereiten. So kann Spanner Grafdurchläufe mit niedriger Latenz ausführen.

Hinweise

  1. Erstellen Sie eine Spanner-Instanz mit einer Datenbank, die Grafdaten enthält. Weitere Informationen finden Sie unter Spanner Graph einrichten und abfragen.

  2. Erstellen Sie in BigQuery eine Slot-Reservierung für die Enterprise- oder Enterprise Plus-Stufe. Sie können die BigQuery-Datenverarbeitungskosten senken, wenn Sie Exporte nach Spanner Graph ausführen. Dazu legen Sie eine Basisslotkapazität von null fest und aktivieren Autoscaling.

  3. Erteilen Sie IAM-Rollen (Identity and Access Management), die Nutzern die erforderlichen Berechtigungen zum Ausführen der einzelnen Aufgaben in diesem Dokument geben.

Erforderliche Rollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen für Ihr Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Exportieren von BigQuery-Graphendaten nach Spanner Graph benötigen:

Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Sie können die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen oder andere vordefinierte Rollen erhalten.

Anwendungsfälle für Reverse ETL

Im Folgenden finden Sie einige Anwendungsbeispiele. Nachdem Sie Daten in BigQuery analysiert und verarbeitet haben, können Sie sie mit Reverse-ETL in Spanner Graph verschieben.

Datenaggregation und -zusammenfassung: Mit BigQuery können Sie Aggregate für detaillierte Daten berechnen, um sie für operative Anwendungsfälle besser geeignet zu machen.

Datentransformation und ‑anreicherung: Mit BigQuery können Sie Daten bereinigen und standardisieren, die aus verschiedenen Datenquellen stammen.

Daten filtern und auswählen: Mit BigQuery können Sie ein großes Dataset für Analysezwecke filtern. So können Sie beispielsweise Daten herausfiltern, die für Echtzeitanwendungen nicht erforderlich sind.

Feature-Vorverarbeitung und ‑Engineering: Verwenden Sie in BigQuery die Funktion ML.TRANSFORM, um Daten zu transformieren, oder die Funktion ML.FEATURE_CROSS, um Feature-Kombinationen von Eingabe-Features zu erstellen. Anschließend verwenden Sie Reverse-ETL, um die resultierenden Daten in Spanner Graph zu verschieben.

Reverse-ETL-Pipeline

Daten werden in einer Reverse-ETL-Pipeline in zwei Schritten von BigQuery in Spanner Graph verschoben:

  1. BigQuery verwendet Slots, die dem Pipelinejob zugewiesen sind, um Quelldaten zu extrahieren und zu transformieren.

  2. Die Reverse-ETL-Pipeline von BigQuery verwendet Spanner-APIs, um Daten in eine bereitgestellte Spanner-Instanz zu laden.

Das folgende Diagramm zeigt die Schritte in einer Reverse-ETL-Pipeline:

Ein Diagramm, das die drei Hauptschritte zeigt, wenn Daten in einer Reverse-ETL-Pipeline von BigQuery nach Spanner Graph übertragen werden.

Abbildung 1. BigQuery-Prozess für Reverse-ETL-Pipelines

Änderungen an Diagrammdaten verwalten

Mit Reverse-ETL können Sie Folgendes tun:

  • Laden Sie ein Graph-Dataset aus BigQuery in Spanner Graph.

  • Spanner Graph-Daten mit laufenden Aktualisierungen aus einem Dataset in BigQuery synchronisieren.

Sie konfigurieren eine Reverse-ETL-Pipeline mit einer SQL-Abfrage, um die Quelldaten und die anzuwendende Transformation anzugeben. Die Pipeline lädt alle Daten, die der WHERE-Klausel der SELECT-Anweisung entsprechen, mit einem Upsert-Vorgang in Spanner. Ein Upsert-Vorgang entspricht INSERT OR UPDATE-Anweisungen. Es werden neue Zeilen eingefügt und vorhandene Zeilen in Tabellen mit Grafdaten aktualisiert. Die Pipeline basiert neue und aktualisierte Zeilen auf einem Spanner-Tabellenprimärschlüssel.

Daten für Tabellen mit Abhängigkeiten von der Ladereihenfolge einfügen und aktualisieren

In den Best Practices für das Schemadesign von Spanner Graph wird empfohlen, verschachtelte Tabellen und Fremdschlüssel zu verwenden. Wenn Sie verschachtelte Tabellen oder erzwungene Fremdschlüssel verwenden, müssen Sie Knoten- und Kantendaten in einer bestimmten Reihenfolge laden. So wird sichergestellt, dass referenzierte Zeilen vorhanden sind, bevor Sie die referenzierende Zeile erstellen. Weitere Informationen finden Sie unter Verschachtelte Tabellen erstellen.

Im folgenden Beispiel für ein Schema für eine Eingabetabelle für einen Graphen werden eine verschränkte Tabelle und eine Fremdschlüsselbeschränkung verwendet, um die Beziehung zwischen einer Person und ihren Konten zu modellieren:

CREATE TABLE Person (
  id    INT64 NOT NULL,
  name  STRING(MAX)
) PRIMARY KEY (id);

CREATE TABLE Account (
  id           INT64 NOT NULL,
  create_time  TIMESTAMP,
  is_blocked   BOOL,
  type        STRING(MAX)
) PRIMARY KEY (id);

CREATE TABLE PersonOwnAccount (
  id           INT64 NOT NULL,
  account_id   INT64 NOT NULL,
  create_time  TIMESTAMP,
  CONSTRAINT FK_Account FOREIGN KEY (account_id) REFERENCES Account (id)
) PRIMARY KEY (id, account_id),
  INTERLEAVE IN PARENT Person ON DELETE CASCADE;

CREATE PROPERTY GRAPH FinGraph
  NODE TABLES (
    Person,
    Account
  )
  EDGE TABLES (
    PersonOwnAccount
      SOURCE KEY (id) REFERENCES Person
      DESTINATION KEY (account_id) REFERENCES Account
      LABEL Owns
  );

In diesem Beispielschema ist PersonOwnAccount eine verschachtelte Tabelle in Person. Laden Sie Elemente in der Tabelle Person vor Elementen in der Tabelle PersonOwnAccount. Außerdem sorgt die Fremdschlüsseleinschränkung für PersonOwnAccount dafür, dass eine übereinstimmende Zeile in Account, dem Ziel der Edge-Beziehung, vorhanden ist. Laden Sie daher die Tabelle Account vor der Tabelle PersonOwnAccount. In der folgenden Liste sind die Abhängigkeiten der Ladereihenfolge dieses Schemas zusammengefasst:

So laden Sie die Daten:

  1. Laden Sie Person vor PersonOwnAccount.
  2. Laden Sie Account vor PersonOwnAccount.

Spanner erzwingt die referenziellen Integritätseinschränkungen im Beispielschema. Wenn die Pipeline versucht, eine Zeile in der Tabelle PersonOwnAccount ohne eine übereinstimmende Zeile in der Tabelle Person oder Account zu erstellen, gibt Spanner einen Fehler zurück. Die Pipeline schlägt dann fehl.

In dieser Reverse-ETL-Pipeline werden EXPORTDATA-Anweisungen in BigQuery verwendet, um Daten aus den Tabellen Person, Account und PersonOwnAccount in einem Dataset zu exportieren, um die Abhängigkeiten der Ladereihenfolge zu erfüllen:

BEGIN
EXPORT DATA OPTIONS (
    uri="https://spanner.googleapis.com/projects/PROJECT_ID/instances/INSTANCE_ID/databases/DATABASE_ID",
    format='CLOUD_SPANNER',
    spanner_options="""{
      "table": "Person",
      "priority": "HIGH",
      "tag" : "graph_data_load_person"
    }"""
  ) AS
  SELECT
    id,
    name
  FROM
    DATASET_NAME.Person;

EXPORT DATA OPTIONS (
  uri="https://spanner.googleapis.com/projects/PROJECT_ID/instances/INSTANCE_ID/databases/DATABASE_ID",
  format='CLOUD_SPANNER',
  spanner_options="""{
    "table": "Account",
    "priority": "HIGH",
    "tag" : "graph_data_load_account"
  }"""
) AS
SELECT
  id,
  create_time,
  is_blocked,
  type
FROM
  DATASET_NAME.Account;

EXPORT DATA OPTIONS (
  uri="https://spanner.googleapis.com/projects/PROJECT_ID/instances/INSTANCE_ID/databases/DATABASE_ID",
  format='CLOUD_SPANNER',
  spanner_options="""{
    "table": "PersonOwnAccount",
    "priority": "HIGH",
    "tag" : "graph_data_load_person_own_account"
  }"""
) AS
SELECT
  id,
  account_id,
  create_time
FROM
  DATASET_NAME.PersonOwnAccount;
END;

Daten synchronisieren

Um BigQuery mit Spanner Graph zu synchronisieren, verwenden Sie Reverse-ETL-Pipelines. Sie können eine Pipeline so konfigurieren, dass sie eine der folgenden Aktionen ausführt:

  • Alle Einfügungen und Aktualisierungen aus der BigQuery-Quelle auf die Spanner Graph-Zieltabelle anwenden. Sie können den Zieltabellen Schemaelemente hinzufügen, um Löschvorgänge logisch zu kommunizieren und Zieltabellenzeilen nach einem Zeitplan zu entfernen.

  • Verwenden Sie eine Zeitreihenfunktion, die Einfüge- und Aktualisierungsvorgänge anwendet und Löschvorgänge erkennt.

Einschränkungen der referenziellen Integrität

Im Gegensatz zu Spanner erzwingt BigQuery keine Einschränkungen für Primär- und Fremdschlüssel. Wenn Ihre BigQuery-Daten nicht den Einschränkungen entsprechen, die Sie für Ihre Spanner-Tabellen erstellen, kann die Reverse-ETL-Pipeline beim Laden dieser Daten fehlschlagen.

Bei Reverse-ETL werden Daten automatisch in Batches gruppiert, die das maximale Mutationslimit pro Commit nicht überschreiten. Die Batches werden atomar und in beliebiger Reihenfolge auf eine Spanner-Tabelle angewendet. Wenn ein Batch Daten enthält, die eine Prüfung der referenziellen Integrität nicht bestehen, wird dieser Batch nicht in Spanner geladen. Beispiele für solche Fehler sind eine verschachtelte untergeordnete Zeile ohne übergeordnete Zeile oder eine erzwungene Spalte mit Fremdschlüssel ohne übereinstimmenden Wert in der referenzierten Spalte. Wenn ein Batch eine Prüfung nicht besteht, schlägt die Pipeline mit einem Fehler fehl und es werden keine weiteren Batches geladen.

Fehler bei der referenziellen Integritätsbeschränkung verstehen

Die folgenden Beispiele zeigen Fehler bei der referenziellen Integrität, die auftreten können:

Fehler bei Fremdschlüsseleinschränkungen beheben

  • Fehler: „Fremdschlüsseleinschränkung FK_Account für Tabelle PersonOwnAccount verletzt. Referenzierte Werte in Account(id) konnten nicht gefunden werden.

  • Ursache: Das Einfügen einer Zeile in die Tabelle PersonOwnAccount ist fehlgeschlagen, weil eine übereinstimmende Zeile in der Tabelle Account fehlt, die für den Fremdschlüssel FK_Account erforderlich ist.

Fehler aufgrund fehlender übergeordneter Zeilen beheben

  • Fehler: „Übergeordnete Zeile für Zeile [15,1] in Tabelle PersonOwnAccount fehlt“

  • Ursache: Das Einfügen einer Zeile in PersonOwnAccount (id: 15 und account_id: 1) ist fehlgeschlagen, weil eine übergeordnete Zeile in der Tabelle Person (id: 15) fehlt.

Um das Risiko von Fehlern bei der referenziellen Integrität zu verringern, sollten Sie die folgenden Optionen in Betracht ziehen. Jede Option hat Vor- und Nachteile.

  • Lockerung der Einschränkungen, damit Spanner Graph Daten laden kann
  • Fügen Sie Ihrer Pipeline Logik hinzu, um Zeilen auszuschließen, die gegen die Einschränkungen der referenziellen Integrität verstoßen.

Referenzielle Integrität lockern

Eine Möglichkeit, Fehler bei der referenziellen Integrität beim Laden von Daten zu vermeiden, besteht darin, die Einschränkungen zu lockern, sodass Spanner die referenzielle Integrität nicht erzwingt.

  • Sie können verschachtelte Tabellen mit der INTERLEAVE IN-Klausel erstellen, um dieselben physischen Zeilenverschachtelungsmerkmale zu verwenden. Wenn Sie INTERLEAVE IN anstelle von INTERLEAVE IN PARENT verwenden, erzwingt Spanner keine referenzielle Integrität. Abfragen profitieren jedoch von der gemeinsamen Speicherung zusammengehöriger Tabellen.

  • Mit der Option NOT ENFORCED können Sie informative Fremdschlüssel erstellen. Die Option NOT ENFORCED bietet Vorteile bei der Abfrageoptimierung. Spanner erzwingt jedoch keine referenzielle Integrität.

Wenn Sie beispielsweise die Edge-Eingabetabelle ohne Prüfungen der referenziellen Integrität erstellen möchten, können Sie diese DDL verwenden:

CREATE TABLE PersonOwnAccount (
  id          INT64 NOT NULL,
  account_id  INT64 NOT NULL,
  create_time TIMESTAMP,
  CONSTRAINT FK_Account FOREIGN KEY (account_id) REFERENCES Account (id) NOT ENFORCED
) PRIMARY KEY (id, account_id),
INTERLEAVE IN Person;

Referenzielle Integrität in Reverse-ETL-Pipelines berücksichtigen

Damit in der Pipeline nur Zeilen geladen werden, die die Prüfungen der referenziellen Integrität bestehen, müssen Sie nur PersonOwnAccount-Zeilen einfügen, die übereinstimmende Zeilen in den Tabellen Person und Account haben. Behalten Sie dann die Ladereihenfolge bei, damit Spanner die Zeilen Person und Account vor den PersonOwnAccount-Zeilen lädt, die auf sie verweisen.

EXPORT DATA OPTIONS (
  uri="https://spanner.googleapis.com/projects/PROJECT_ID/instances/INSTANCE_ID/databases/DATABASE_ID",
  format='CLOUD_SPANNER',
    spanner_options="""{
      "table": "PersonOwnAccount",
      "priority": "HIGH",
      "tag" : "graph_data_load_person_own_account"
    }"""
  ) AS
  SELECT
    poa.id,
    poa.account_id,
    poa.create_time
  FROM `PROJECT_ID.DATASET_NAME.PersonOwnAccount` poa
    JOIN `PROJECT_ID.DATASET_NAME.Person` p ON (poa.id = p.id)
    JOIN `PROJECT_ID.DATASET_NAME.Account` a ON (poa.account_id = a.id)
  WHERE poa.id = p.id
    AND poa.account_id = a.id;

Graphelemente löschen

Für Reverse-ETL-Pipelines werden Upsert-Vorgänge verwendet. Da Upsert-Vorgänge INSERT OR UPDATE-Anweisungen entsprechen, kann eine Pipeline nur Zeilen synchronisieren, die zur Laufzeit in den Quelldaten vorhanden sind. Das bedeutet, dass gelöschte Zeilen in der Pipeline ausgeschlossen werden. Wenn Sie Daten aus BigQuery löschen, kann eine Reverse-ETL-Pipeline dieselben Daten nicht direkt aus Spanner Graph entfernen.

Sie haben folgende Möglichkeiten, um Löschungen aus BigQuery-Quelltabelle zu verarbeiten:

Logisches oder vorläufiges Löschen in der Quelle ausführen

Verwenden Sie ein gelöschtes Flag in BigQuery, um Zeilen logisch zum Löschen zu markieren. Erstellen Sie dann eine Spalte in der Ziel-Spanner-Tabelle, in die Sie das Flag übertragen können. Wenn das Reverse-ETL die Pipeline-Aktualisierungen anwendet, löschen Sie Zeilen mit diesem Flag in Spanner. Sie können solche Zeilen explizit mit partitionierter DML suchen und löschen. Alternativ können Sie Zeilen implizit löschen, indem Sie eine TTL-Spalte (Time to Live) mit einem Datum konfigurieren, das von der Spalte mit dem Löschflag abhängt. Schreiben Sie Spanner-Abfragen, um diese logisch gelöschten Zeilen auszuschließen. So wird sichergestellt, dass Spanner diese Zeilen vor dem geplanten Löschen aus den Ergebnissen ausschließt. Nachdem die Reverse-ETL-Pipeline vollständig ausgeführt wurde, werden die logischen Löschvorgänge in den Spanner-Zeilen widergespiegelt. Anschließend können Sie Zeilen aus BigQuery löschen.

In diesem Beispiel wird der Tabelle PersonOwnAccount in Spanner die Spalte is_deleted hinzugefügt. Anschließend wird eine Spalte expired_ts_generated hinzugefügt, die vom Wert is_deleted abhängt. Durch die TTL-Richtlinie werden die betroffenen Zeilen zum Löschen geplant, da das Datum in der generierten Spalte vor dem DELETION POLICY-Schwellenwert liegt.

ALTER TABLE PersonOwnAccount
  ADD COLUMN is_deleted BOOL DEFAULT (FALSE);

ALTER TABLE PersonOwnAccount ADD COLUMN
  expired_ts_generated TIMESTAMP AS (IF(is_deleted,
    TIMESTAMP("1970-01-01 00:00:00+00"),
    TIMESTAMP("9999-01-01 00:00:00+00"))) STORED HIDDEN;

ALTER TABLE PersonOwnAccount
  ADD ROW DELETION POLICY (OLDER_THAN(expired_ts_generated, INTERVAL 0 DAY));

BigQuery-Änderungsverlauf für INSERT-, UPDATE- und logische Löschvorgänge verwenden

Sie können Änderungen an einer BigQuery-Tabelle anhand ihres Änderungsverlaufs nachvollziehen. Mit der GoogleSQL-Funktion CHANGES können Sie nach Zeilen suchen, die sich innerhalb eines bestimmten Zeitintervalls geändert haben. Verwenden Sie dann die Informationen zur gelöschten Zeile mit einer Reverse-ETL-Pipeline. Sie können die Pipeline so einrichten, dass in der Spanner-Tabelle ein Indikator wie ein gelöschtes Flag oder ein Ablaufdatum festgelegt wird. Mit diesem Indikator werden Zeilen in den Spanner-Tabellen zum Löschen markiert.

Anhand der Ergebnisse der Zeitachsenfunktion CHANGES können Sie entscheiden, welche Zeilen aus der Quelltabelle in den Load Ihrer Reverse-ETL-Pipeline aufgenommen werden sollen.

Die Pipeline enthält Zeilen mit _CHANGE_TYPE als INSERT oder UPDATE als Upserts, wenn die Zeile in der Quelltabelle vorhanden ist. Die aktuelle Zeile aus der Quelltabelle enthält die neuesten Daten.

Verwenden Sie Zeilen mit _CHANGE_TYPE als DELETE, die keine vorhandenen Zeilen in der Quelltabelle haben, um einen Indikator in der Spanner-Tabelle festzulegen, z. B. ein gelöschtes Flag oder ein Ablaufdatum für die Zeile.

In Ihrer Exportabfrage muss die Reihenfolge der Einfügungen und Löschungen in BigQuery berücksichtigt werden. Stellen Sie sich beispielsweise vor, dass eine Zeile zum Zeitpunkt T1 gelöscht und eine neue Zeile zu einem späteren Zeitpunkt T2 eingefügt wird. Wenn beide derselben Spanner-Tabellenzeile zugeordnet werden, müssen die Auswirkungen dieser Ereignisse beim Export in ihrer ursprünglichen Reihenfolge beibehalten werden.

Wenn festgelegt, markiert der Indikator delete Zeilen in den Spanner-Tabellen zum Löschen.

Sie können beispielsweise einer Spanner-Eingabetabelle eine Spalte hinzufügen, um das Ablaufdatum jeder Zeile zu speichern. Erstellen Sie dann eine Löschrichtlinie, in der diese Ablaufdaten verwendet werden.

Im folgenden Beispiel wird gezeigt, wie Sie eine Spalte hinzufügen, in der die Ablaufdaten der Zeilen der Tabelle gespeichert werden.

ALTER TABLE PersonOwnAccount ADD COLUMN expired_ts TIMESTAMP;

ALTER TABLE PersonOwnAccount
  ADD ROW DELETION POLICY (OLDER_THAN(expired_ts, INTERVAL 1 DAY));

Wenn Sie die Funktion CHANGES für eine Tabelle in BigQuery verwenden möchten, legen Sie die Option enable_change_history der Tabelle auf TRUE fest:

ALTER TABLE `PROJECT_ID.DATASET_NAME.PersonOwnAccount`
  SET OPTIONS (enable_change_history=TRUE);

Im folgenden Beispiel wird gezeigt, wie Sie mit Reverse-ETL neue oder geänderte Zeilen aktualisieren und das Ablaufdatum für zum Löschen markierte Zeilen festlegen können. Durch einen LEFT JOIN mit der Tabelle PersonOwnAccount erhält die Abfrage Informationen zum aktuellen Status jeder Zeile.

EXPORT DATA OPTIONS (
  uri="https://spanner.googleapis.com/projects/PROJECT_ID/instances/INSTANCE_ID/databases/DATABASE_ID",
    format='CLOUD_SPANNER',
    spanner_options="""{
      "table": "PersonOwnAccount",
      "priority": "HIGH",
      "tag" : "graph_data_delete_via_reverse_etl"
    }"""
  ) AS
SELECT
  DISTINCT
   IF (changes._CHANGE_TYPE = 'DELETE', changes.id, poa.id) AS id,
   IF (changes._CHANGE_TYPE = 'DELETE', changes.account_id, poa.account_id) AS account_id,
   IF (changes._CHANGE_TYPE = 'DELETE', changes.create_time, poa.create_time) AS create_time,
   IF (changes._CHANGE_TYPE = 'DELETE', changes._CHANGE_TIMESTAMP, NULL) AS expired_ts
FROM
  CHANGES(TABLE `PROJECT_ID.DATASET_NAME.PersonOwnAccount`,
    TIMESTAMP_TRUNC(TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY), DAY),
    TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), DAY)) changes
LEFT JOIN `PROJECT_ID.DATASET_NAME.PersonOwnAccount` poa
  ON (poa.id = changes.id
  AND poa.account_id = changes.account_id)
WHERE (changes._CHANGE_TYPE = 'DELETE'
   AND poa.id IS NULL)
   OR (changes._CHANGE_TYPE IN ( 'UPDATE', 'INSERT')
   AND poa.id IS NOT NULL );

In der Beispielabfrage wird ein LEFT JOIN mit der Quelltabelle verwendet, um die Reihenfolge beizubehalten. Durch diesen Join werden DELETE-Änderungsdatensätze für Zeilen ignoriert, die im Änderungsverlaufszeitraum der Abfrage gelöscht und dann neu erstellt wurden. Die Pipeline behält die gültige neue Zeile bei.

Wenn Sie Zeilen löschen, wird in der Pipeline die Spalte expired_ts in der entsprechenden Spanner Graph-Zeile mit dem DELETE-Zeitstempel aus der Spalte _CHANGE_TIMESTAMP gefüllt. Eine Richtlinie zum Löschen von Zeilen (TTL-Richtlinie) in Spanner löscht alle Zeilen, in denen der Wert expired_ts mehr als einen Tag in der Vergangenheit liegt.

Um die Systemzuverlässigkeit zu gewährleisten, sollten Sie den Zeitplan der Pipeline, das Lookback-Fenster für Änderungen und die Spanner-TTL-Richtlinie aufeinander abstimmen. Planen Sie die tägliche Ausführung der Pipeline. Die Spanner-TTL-Richtlinie muss eine Dauer haben, die länger als dieses Ausführungsintervall ist. Dadurch wird verhindert, dass die Pipeline ein vorheriges DELETE-Ereignis für eine Zeile noch einmal verarbeitet, die bereits durch die Spanner-TTL-Richtlinie entfernt wurde.

In diesem Beispiel wird das Intervall start_timestamp und end_timestamp für tägliche Abfragen gezeigt, in denen alle Änderungen an der BigQuery-Tabelle vom vorherigen UTC-Tag erfasst werden. Da es sich um eine Batchabfrage handelt und die Funktion CHANGES Einschränkungen unterliegt, muss end_timestamp mindestens 10 Minuten vor der aktuellen Zeit liegen. Planen Sie daher die Ausführung dieser Abfrage mindestens 10 Minuten nach Mitternacht UTC. Weitere Informationen finden Sie in der Dokumentation zu CHANGES.

TTL-Spalten mit Zeitstempel der letzten Sichtung verwenden

In einer Reverse-ETL-Pipeline wird für jede Zeile in der Spanner-Tabelle die Spalte last_seen_ts auf den aktuellen Zeitstempel gesetzt. Wenn Sie BigQuery-Zeilen löschen, werden die entsprechenden Zeilen in Spanner nicht aktualisiert und die Spalte last_seen_ts ändert sich nicht. Spanner entfernt dann Zeilen mit einem veralteten last_seen_ts mithilfe einer TTL-Richtlinie oder partitionierten DML basierend auf einem definierten Schwellenwert. Vor dem geplanten Löschen können Spanner-Abfragen Zeilen mit einem last_seen_ts filtern, das älter als dieser Schwellenwert ist. Dieser Ansatz funktioniert effektiv, wenn die Grafiken regelmäßig aktualisiert werden und fehlende Aktualisierungen auf veraltete Daten hinweisen, die gelöscht werden müssen.

Vollständige Aktualisierung durchführen

Vor dem Laden aus BigQuery können Sie Spanner-Tabellen löschen, um Löschvorgänge in den Quelltabellen zu berücksichtigen. So wird verhindert, dass bei der nächsten Ausführung der Pipeline Zeilen, die aus den BigQuery-Quelltabelle gelöscht wurden, in Spanner geladen werden. Dies ist möglicherweise die einfachste Option für die Implementierung. Berücksichtigen Sie jedoch die Zeit, die zum vollständigen Neuladen Ihrer Grafiken erforderlich ist.

Geplante Batch-Reverse-ETL-Pipeline verwalten

Nachdem bei der ersten Ausführung Ihrer Reverse-ETL-Pipeline Daten aus BigQuery in Spanner Graph bulk-geladen wurden, ändern sich die Daten in der realen Welt ständig. Datasets ändern sich und die Pipeline fügt im Laufe der Zeit Grafikelemente hinzu oder entfernt sie. In der Pipeline werden neue Knoten erkannt und neue Edge-Beziehungen hinzugefügt oder durch KI-Inferenz generiert.

Damit die Spanner Graph-Datenbank auf dem neuesten Stand bleibt, können Sie die BigQuery-Pipeline-Orchestrierung planen und sequenzieren. Verwenden Sie dazu eine der folgenden Optionen:

Mit BigQuery-Pipelines können Sie komplexe SQL-Workflows für die Datentransformation in BigQuery entwickeln, testen, versionieren und bereitstellen. Außerdem werden Abhängigkeiten zwischen den einzelnen Schritten berücksichtigt, da Sie Beziehungen zwischen den Abfragen in Ihrer Pipeline definieren können. Dataform erstellt einen Abhängigkeitsbaum und führt Ihre Abfragen in der richtigen Reihenfolge aus. So wird sichergestellt, dass Upstream-Abhängigkeiten abgeschlossen werden, bevor Downstream-Aufgaben beginnen.

Workflows, die von Cloud Scheduler aufgerufen werden, bieten eine nützliche und flexible Lösung zum Orchestrieren von Sequenzen vonGoogle Cloud -Diensten, einschließlich BigQuery-Abfragen. Ein Workflow wird als eine Reihe von Schritten definiert, in denen jeweils ein BigQuery-Job ausgeführt wird. Mit Cloud Scheduler können Sie diese Workflows nach einem definierten Zeitplan aufrufen. Sie können Abhängigkeiten mithilfe der Workflowdefinition verwalten, um die Ausführungsreihenfolge festzulegen, bedingte Logik zu implementieren, Fehler zu beheben und Ausgaben von einer Abfrage an eine andere zu übergeben.

Mit geplanten Abfragen, auch als BigQuery-Übertragungsjobs bezeichnet, können Sie SQL-Anweisungen in BigQuery regelmäßig ausführen. Geplante Abfragen bieten keine robuste Fehlerbehandlung oder dynamische Abhängigkeitsverwaltung.

Umgekehrtes ETL mit kontinuierlichen BigQuery-Abfragen

Mit der Funktion kontinuierliche BigQuery-Abfragen können Sie BigQuery-Vorgänge nahezu in Echtzeit ausführen. Durch die Kombination von EXPORT DATA mit kontinuierlichen Abfragen wird eine alternative Methode zum Ausführen von Reverse-ETL-Pipelines bereitgestellt, bei der geplante Batchjobs vermieden werden.

Eine kontinuierliche Abfrage ist eine lang andauernde Abfrage, mit der eine BigQuery-Quelltabelle auf neue Zeilen überwacht wird. Wenn BigQuery erkennt, dass der Tabelle neue Zeilen angehängt wurden, werden die Abfrageergebnisse an den EXPORT DATA-Vorgang gestreamt.

Dieser Ansatz bietet folgende Vorteile:

  • Datensynchronisierung nahezu in Echtzeit: Neue Zeilen in BigQuery werden mit minimaler Verzögerung in Spanner übernommen.

  • Geringerer Overhead für die Batchverarbeitung: Durch eine kontinuierliche Abfrage sind keine regelmäßigen Batchjobs mehr erforderlich, was den Rechenaufwand reduziert.

  • Ereignisgesteuerte Aktualisierungen: Spanner-Daten werden als Reaktion auf tatsächliche Änderungen in BigQuery aktualisiert.

Für eine Pipeline für kontinuierliche Abfragen ist eine Slot-Reservierungszuweisung mit dem job_type CONTINUOUS erforderlich. Weisen Sie diese auf Projekt- oder Ordnerebene oder auf Organisationsebene zu.

Continuous Query mit umgekehrtem ETL von BigQuery zu Spanner erstellen

Konfigurieren Sie den Parameter start_timestamp der Funktion APPENDS, um mit der Verarbeitung von Daten dort zu beginnen, wo der Batch-Ladevorgang unterbrochen wurde. Mit dieser Funktion werden alle Zeilen erfasst, die im angegebenen Zeitraum erstellt wurden. Im folgenden Beispiel wird der Startpunkt der Pipeline willkürlich auf 10 Minuten vor dem CURRENT_TIME festgelegt. Dieser Zeitstempel muss innerhalb des BigQuery-Zeitreisefensters liegen.

Es gibt mehrere Methoden, um eine Pipeline für kontinuierliche Abfragen zu starten, darunter die folgenden:

  1. Wählen Sie in BigQuery Studio Mehr und dann unter Abfragemodus auswählen die Option Continuous Query aus.

  2. Verwenden Sie die bq-Befehlszeile und geben Sie die Option --continuous=true an.

EXPORT DATA OPTIONS ( uri="https://spanner.googleapis.com/projects/PROJECT_ID/instances/INSTANCE_ID/databases/DATABASE_ID",
  format="CLOUD_SPANNER",
  spanner_options="""{
      "table": "PersonOwnAccount",
      "priority": "HIGH",
      "tag": "reverse-etl-continuous",
      "change_timestamp_column": "create_time"
   }"""
)
AS SELECT id, account_id, _CHANGE_TIMESTAMP as create_time
  FROM
APPENDS(TABLE `PROJECT_ID.DATASET_NAME.PersonOwnAccount`,
  CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE )

Reihenfolge des Ladens nicht garantiert

Spanner Graph-Daten bestehen aus mehreren Eingabetabellen. Sie müssen eine strikte Ladereihenfolge einhalten, wenn für Tabellen Einschränkungen der referenziellen Integrität gelten. Bei gleichzeitigen Continuous Queries kann jedoch nicht gesteuert werden, in welcher Reihenfolge Spanner Zeilen hinzufügt. Daher ist das Laden von Spanner Graph-Daten mit kontinuierlichen Abfragen nur für Graphschemata mit gelockerten Einschränkungen der referenziellen Integrität geeignet.

In vorhandene Pipelines einbinden

Continuous Querys ergänzen vorhandene geplante Batchjobs. Verwenden Sie beispielsweise kontinuierliche Abfragen für Aktualisierungen nahezu in Echtzeit und geplante Jobs für die vollständige Datensynchronisierung oder -abstimmung.

Mit kontinuierlichen BigQuery-Abfragen können Sie reaktionsschnelle und aktuelle Reverse-ETL-Pipelines erstellen, um Daten zwischen BigQuery und Spanner Graph zu synchronisieren.

Überlegungen zu kontinuierlichen Abfragen

  • Kosten: Für kontinuierliche Abfragen fallen Kosten für die laufende Ausführung von Abfragen und das Streamen von Daten an.

  • Fehlerbehandlung: Eine Pipeline für kontinuierliche Abfragen wird abgebrochen, wenn Datenbankfehler auftreten, z. B. ein doppelter Primärschlüssel oder ein Verstoß gegen die referenzielle Integrität. Wenn eine Pipeline fehlschlägt, müssen Sie die Daten in der BigQuery-Quelltabelle manuell korrigieren, bevor Sie die Abfrage neu starten.

  • Löschungen und Aktualisierungen werden nicht berücksichtigt: Mit der Funktion APPENDS werden nur Einfügungen erfasst. Löschungen oder Aktualisierungen werden nicht erfasst.

Best Practices für Reverse-ETL befolgen

Für optimale Ergebnisse sollten Sie Folgendes beachten.

  • Wählen Sie eine Strategie aus, um Fehler bei der referenziellen Integrität zu vermeiden, wenn Sie Kantendaten laden.

  • Entwerfen Sie Ihre gesamte Datenpipeline so, dass keine losen Kanten entstehen. Lose Kanten können die Effizienz von Spanner Graph-Abfragen und die Integrität der Graphstruktur beeinträchtigen. Weitere Informationen finden Sie unter Lose Kanten verhindern.

  • Befolgen Sie die Empfehlungen zur Exportoptimierung für Spanner.

  • Wenn Sie eine große Datenmenge laden, sollten Sie die Pipeline in mehrere kleinere Pipelines aufteilen, um das standardmäßige Zeitkontingent von sechs Stunden für die Ausführung von BigQuery-Abfragen nicht zu überschreiten. Weitere Informationen finden Sie unter Limits für BigQuery-Abfragejobs.

  • Bei großen Datenmengen sollten Sie Indexe und Fremdschlüssel-Einschränkungen erst nach dem ersten Massenladen der Daten hinzufügen. Diese Vorgehensweise verbessert die Leistung beim Laden von Daten, da für Fremdschlüsseleinschränkungen zusätzliche Lesevorgänge zur Validierung und für Indexe zusätzliche Schreibvorgänge erforderlich sind. Diese Vorgänge erhöhen die Anzahl der Transaktionsbeteiligten, was den Datenladevorgang verlangsamen kann.

  • Aktivieren Sie das Autoscaling in Spanner, um die Ladezeiten von Daten in eine Instanz zu verkürzen. Konfigurieren Sie dann den Spanner-Parameter priority im Abschnitt spanner_options des BigQuery-Befehls EXPORT DATA auf HIGH. Weitere Informationen finden Sie unter Übersicht über das Autoscaling von Spanner, Exporte mit der Option spanner_options konfigurieren und RequestOptions.priority.

  • Bei großen Datenmengen sollten Sie Aufteilungspunkte erstellen, um Ihre Datenbank vorab aufzuteilen. Dadurch wird Spanner auf einen erhöhten Durchsatz vorbereitet.

  • Konfigurieren Sie die Anfragepriorität für Spanner für das Laden von Daten in der Pipelinedefinition.

Nächste Schritte