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 in 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-Wert und eine niedrige Bereitstellungs-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.

Hinweis

  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-Version. 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 BigQuery Reverse-ETL-Pipeline 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

Bei Best Practices für das Spanner Graph-Schemadesign 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 Graph-Eingabetabellenschema 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": "LOW",
      "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": "LOW",
    "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": "LOW",
    "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 bei einer Prüfung der referenziellen Integrität fehlschlagen, 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ät verstehen

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

Fehler bei Fremdschlüsselbeschränkungen beheben

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

  • 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.

  • Lockeren Sie die Einschränkungen, damit Spanner Graph Daten laden kann.
  • Fügen Sie Ihrer Pipeline Logik hinzu, um Zeilen auszulassen, 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 so zu lockern, dass 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 verwandter 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 mit entsprechenden Zeilen in den Tabellen Person und Account einfügen. 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": "LOW",
      "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;

Grafikelemente löschen

Reverse-ETL-Pipelines verwenden Upsert-Vorgänge. 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 des Ä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 Zeitreihenfunktion CHANGES können Sie entscheiden, welche Zeilen aus der Quelltabelle in den Ladevorgang 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 Richtlinie zur Datenlöschung, 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/