Daten mit Dataflow importieren, exportieren und ändern

Auf dieser Seite wird beschrieben, wie Sie mit dem Dataflow-Connector für Spanner Daten in Spanner-Datenbanken mit GoogleSQL-Dialekt und PostgreSQL-Dialekt importieren, exportieren und ändern.

Dataflow ist ein verwalteter Dienst für die Transformation und Anreicherung von Daten. Mit dem Dataflow-Connector für Spanner können Sie über eine Dataflow-Pipeline Daten in Spanner lesen und schreiben. Optional können Sie die Daten dabei transformieren oder ändern. Sie können auch Pipelines erstellen, mit denen Daten zwischen Spanner und anderen Google Cloud Produkten übertragen werden.

Der Dataflow-Connector ist die empfohlene Methode zur effizienten Bulk-Verschiebung von Daten in und aus Spanner. Er ist auch die empfohlene Methode für große Transformationen an einer Datenbank, die von Partitioned DML nicht unterstützt werden, z. B. Tabellenverschiebungen und Bulk-Löschvorgänge, für die ein JOIN erforderlich ist. Wenn Sie mit einzelnen Datenbanken arbeiten, können Sie die Daten mit anderen Methoden importieren und exportieren:

  • Verwenden Sie die Google Cloud console, um eine einzelne Datenbank im Avro Format aus Spanner nach Cloud Storage zu exportieren.
  • Verwenden Sie die Google Cloud console, um eine Datenbank aus Dateien, die Sie nach Cloud Storage exportiert haben, zurück in Spanner zu importieren.
  • Verwenden Sie die REST API oder die Google Cloud CLI, um Export oder Import Jobs von Spanner nach Cloud Storage und zurück auszuführen. Dabei können Sie auch das Avro-Format verwenden.

Der Dataflow-Connector für Spanner ist Teil des Apache Beam Java SDK und bietet eine API für die oben genannten Vorgänge. Weitere Informationen zu einigen der auf dieser Seite erläuterten Konzepte, wie PCollection Objekte und Transformationen, finden Sie in der Apache Beam-Programmier anleitung.

Connector dem Maven-Projekt hinzufügen

Wenn Sie den Google Cloud Dataflow-Connector einem Maven Projekt hinzufügen möchten, fügen Sie das beam-sdks-java-io-google-cloud-platform Maven-Artefakt als Abhängigkeit in die Datei pom.xml ein.

Angenommen, in der Datei pom.xml ist beam.version auf die entsprechende Versionsnummer festgelegt. In diesem Fall fügen Sie die folgende Abhängigkeit hinzu:

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>${beam.version}</version>
</dependency>

Daten aus Spanner lesen

Wenn Sie Daten aus Spanner lesen möchten, wenden Sie die SpannerIO.read Transformation an. Konfigurieren Sie den Lesevorgang mit den Methoden in der SpannerIO.Read Klasse. Durch Einsatz der Transformation wird das Objekt PCollection<Struct> zurückgegeben. Dabei stellt jedes Element in der Sammlung eine einzelne Zeile dar, die vom Lesevorgang zurückgegeben wird. Je nach gewünschter Ausgabe können Sie Daten mit einer bestimmten SQL-Abfrage oder ohne SQL-Abfrage aus Spanner lesen.

Durch Einsatz der Transformation SpannerIO.read wird mithilfe eines starken Lesevorgangs eine konsistente Datenansicht zurückgegeben. Sofern Sie dies nicht anders festlegen, beruht das Ergebnis des Lesevorgangs auf einem Snapshot der Daten zu dem Zeitpunkt, an dem Sie den Lesevorgang gestartet haben. Weitere Informationen zu den verschiedenen Arten von Lesevorgängen, die mit Spanner möglich sind, finden Sie unter Lesevorgänge.

Daten mit einer Abfrage lesen

Wenn Sie bestimmte Daten aus Spanner lesen möchten, konfigurieren Sie die Transformation mit der SpannerIO.Read.withQuery Methode, um eine SQL-Abfrage anzugeben. Beispiel:

// Query for all the columns and rows in the specified Spanner table
PCollection<Struct> records = pipeline.apply(
    SpannerIO.read()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withQuery("SELECT * FROM " + options.getTable()));

Daten lesen, ohne eine Abfrage anzugeben

Wenn Sie Daten aus einer Datenbank lesen möchten, ohne eine Abfrage zu verwenden, können Sie mit der SpannerIO.Read.withTable Methode einen Tabellen namen und mit der SpannerIO.Read.withColumns Methode eine Liste der zu lesenden Spalten angeben. Beispiel:

GoogleSQL 

// Query for all the columns and rows in the specified Spanner table
PCollection<Struct> records = pipeline.apply(
    SpannerIO.read()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withTable("Singers")
        .withColumns("singerId", "firstName", "lastName"));

PostgreSQL 

// Query for all the columns and rows in the specified Spanner table
PCollection<Struct> records = pipeline.apply(
    SpannerIO.read()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withTable("singers")
        .withColumns("singer_id", "first_name", "last_name"));

Wenn Sie die Anzahl der gelesenen Zeilen begrenzen möchten, können Sie mit der SpannerIO.Read.withKeySet Methode eine Reihe von Primärschlüsseln angeben, die gelesen werden sollen.

Sie können eine Tabelle auch mit einem angegebenen sekundären Index lesen. Wie beim readUsingIndex API-Aufruf, muss der Index alle Daten enthalten, die in den Abfrageergebnissen angezeigt werden.

Geben Sie dazu die Tabelle wie im vorherigen Beispiel an und geben Sie mit der SpannerIO.Read.withIndex Methode den Index an, der die erforderlichen Spaltenwerte enthält. Der Index muss alle Spalten speichern, die von der Transformation gelesen werden müssen. Der Primärschlüssel der Basistabelle wird implizit gespeichert. Wenn Sie beispielsweise die Tabelle Songs mit dem Index SongsBySongName lesen möchten, verwenden Sie den folgenden Code:

GoogleSQL 

// Read the indexed columns from all rows in the specified index.
PCollection<Struct> records =
    pipeline.apply(
        SpannerIO.read()
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId)
            .withTable("Songs")
            .withIndex("SongsBySongName")
            // Can only read columns that are either indexed, STORED in the index or
            // part of the primary key of the Songs table,
            .withColumns("SingerId", "AlbumId", "TrackId", "SongName"));

PostgreSQL 

// // Read the indexed columns from all rows in the specified index.
PCollection<Struct> records =
    pipeline.apply(
        SpannerIO.read()
            .withInstanceId(instanceId)
            .withDatabaseId(databaseId)
            .withTable("Songs")
            .withIndex("SongsBySongName")
            // Can only read columns that are either indexed, STORED in the index or
            // part of the primary key of the songs table,
            .withColumns("singer_id", "album_id", "track_id", "song_name"));

Veralterung von Transaktionsdaten steuern

Eine Transformation wird immer auf einen konsistenten Daten-Snapshot angewendet. Verwenden Sie die Methode SpannerIO.Read.withTimestampBound, um die Veralterung der Daten zu steuern. Weitere Informationen finden Sie unter Transaktionen.

Lesen aus mehreren Tabellen mit der gleichen Transaktion

Wenn Sie Daten aus mehreren Tabellen zur gleichen Zeit lesen möchten, damit Sie so die Datenkonsistenz gewährleisten können, führen Sie alle Lesevorgänge mit einer einzigen Transaktion aus. Wenden Sie dazu eine createTransaction-Transformation an, um ein PCollectionView<Transaction>-Objekt zu erstellen, mit dem dann eine Transaktion erstellt wird. Die entstehende Ansicht kann mit SpannerIO.Read.withTransaction an einen Lesevorgang übergeben werden.

GoogleSQL 

SpannerConfig spannerConfig =
    SpannerConfig.create().withInstanceId(instanceId).withDatabaseId(databaseId);
PCollectionView<Transaction> tx =
    pipeline.apply(
        SpannerIO.createTransaction()
            .withSpannerConfig(spannerConfig)
            .withTimestampBound(TimestampBound.strong()));
PCollection<Struct> singers =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT SingerID, FirstName, LastName FROM Singers")
            .withTransaction(tx));
PCollection<Struct> albums =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT SingerId, AlbumId, AlbumTitle FROM Albums")
            .withTransaction(tx));

PostgreSQL 

SpannerConfig spannerConfig =
    SpannerConfig.create().withInstanceId(instanceId).withDatabaseId(databaseId);
PCollectionView<Transaction> tx =
    pipeline.apply(
        SpannerIO.createTransaction()
            .withSpannerConfig(spannerConfig)
            .withTimestampBound(TimestampBound.strong()));
PCollection<Struct> singers =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT singer_id, first_name, last_name FROM singers")
            .withTransaction(tx));
PCollection<Struct> albums =
    pipeline.apply(
        SpannerIO.read()
            .withSpannerConfig(spannerConfig)
            .withQuery("SELECT singer_id, album_id, album_title FROM albums")
            .withTransaction(tx));

Daten aus allen verfügbaren Tabellen lesen

Sie können die Daten aus allen verfügbaren Tabellen in einer Spanner-Datenbank lesen.

GoogleSQL 

PCollection<Struct> allRecords =
    pipeline
        .apply(
            SpannerIO.read()
                .withSpannerConfig(spannerConfig)
                .withBatching(false)
                .withQuery(
                    "SELECT t.table_name FROM information_schema.tables AS t WHERE t"
                        + ".table_catalog = '' AND t.table_schema = ''"))
        .apply(
            MapElements.into(TypeDescriptor.of(ReadOperation.class))
                .via(
                    (SerializableFunction<Struct, ReadOperation>)
                        input -> {
                          String tableName = input.getString(0);
                          return ReadOperation.create().withQuery("SELECT * FROM " + tableName);
                        }))
        .apply(SpannerIO.readAll().withSpannerConfig(spannerConfig));

PostgreSQL 

PCollection<Struct> allRecords =
    pipeline
        .apply(
            SpannerIO.read()
                .withSpannerConfig(spannerConfig)
                .withBatching(false)
                .withQuery(
                    Statement.newBuilder(
                            "SELECT t.table_name FROM information_schema.tables AS t "
                                + "WHERE t.table_catalog = $1 AND t.table_schema = $2")
                        .bind("p1")
                        .to(spannerConfig.getDatabaseId().get())
                        .bind("p2")
                        .to("public")
                        .build()))
        .apply(
            MapElements.into(TypeDescriptor.of(ReadOperation.class))
                .via(
                    (SerializableFunction<Struct, ReadOperation>)
                        input -> {
                          String tableName = input.getString(0);
                          return ReadOperation.create()
                              .withQuery("SELECT * FROM \"" + tableName + "\"");
                        }))
        .apply(SpannerIO.readAll().withSpannerConfig(spannerConfig));

Fehlerbehebung bei nicht unterstützten Abfragen

Der Dataflow-Connector unterstützt nur SQL -Abfragen an Spanner, bei denen der erste Operator im Abfrageausführungsplan Distributed Union ist. Wenn Sie versuchen, Daten aus Spanner mit einer Abfrage zu lesen und die Ausnahmemeldung erhalten, dass die Abfrage does not have a DistributedUnion at the root, führen Sie die unter So werden Abfragen von Spanner ausgeführt angegebenen Schritte aus, damit Sie mit der Google Cloud console einen Ausführungsplan für die Abfrage abrufen können.

Wenn die SQL-Abfrage nicht unterstützt wird, vereinfachen Sie sie entsprechend, damit sie als ersten Operator im Ausführungsplan den Operator "Distributed Union" enthält. Entfernen Sie Aggregatfunktionen, Tabellen-Joins sowie die Operatoren DISTINCT, GROUP BY und ORDER, da diese Operatoren am ehesten verhindern, dass die Abfrage funktioniert.

Mutationen für einen Schreibvorgang erstellen

Verwenden Sie die Mutation Klasse's newInsertOrUpdateBuilder Methode anstelle der newInsertBuilder Methode , sofern dies für Java-Pipelines nicht unbedingt erforderlich ist. Verwenden Sie für Python-Pipelines SpannerInsertOrUpdate anstelle von SpannerInsert. Dataflow garantiert die mindestens einmalige Ausführung. Dies bedeutet, dass die Mutation möglicherweise mehrmals geschrieben wird. Daher können Mutationen vom Typ INSERT Fehler vom Typ com.google.cloud.spanner.SpannerException: ALREADY_EXISTS verursachen, die zu einem Ausfall der Pipeline führen. Um diesen Fehler zu vermeiden, verwenden Sie stattdessen die Mutation INSERT_OR_UPDATE. Damit wird eine neue Zeile hinzugefügt oder Spaltenwerte aktualisiert, falls die Zeile bereits vorhanden ist. Die Mutation INSERT_OR_UPDATE kann mehrmals angewendet werden.

In Spanner schreiben und Daten transformieren

Mit dem Dataflow Connector können Sie Daten in Spanner schreiben. Verwenden Sie hierfür die Transformation SpannerIO.write, um eine Sammlung von Eingabezeilenmutationen auszuführen. Für eine höhere Effizienz gruppiert der Dataflow-Connector Mutationen in Batches.

Das folgende Beispiel zeigt, wie eine Schreibtransformation auf eine PCollection von Mutationen angewendet wird:

GoogleSQL 

albums
    // Spanner expects a Mutation object, so create it using the Album's data
    .apply("CreateAlbumMutation", ParDo.of(new DoFn<Album, Mutation>() {
      @ProcessElement
      public void processElement(ProcessContext c) {
        Album album = c.element();
        c.output(Mutation.newInsertOrUpdateBuilder("albums")
            .set("singerId").to(album.singerId)
            .set("albumId").to(album.albumId)
            .set("albumTitle").to(album.albumTitle)
            .build());
      }
    }))
    // Write mutations to Spanner
    .apply("WriteAlbums", SpannerIO.write()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId));

PostgreSQL 

PCollectionView<Dialect> dialectView =
    pipeline.apply(Create.of(Dialect.POSTGRESQL)).apply(View.asSingleton());
albums
    // Spanner expects a Mutation object, so create it using the Album's data
    .apply("CreateAlbumMutation", ParDo.of(new DoFn<Album, Mutation>() {
      @ProcessElement
      public void processElement(ProcessContext c) {
        Album album = c.element();
        c.output(Mutation.newInsertOrUpdateBuilder("albums")
            .set("singerId").to(album.singerId)
            .set("albumId").to(album.albumId)
            .set("albumTitle").to(album.albumTitle)
            .build());
      }
    }))
    // Write mutations to Spanner
    .apply("WriteAlbums", SpannerIO.write()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withDialectView(dialectView));

Wenn eine Transformation unerwartet vor Abschluss beendet wird, werden bereits angewendete Mutationen nicht zurückgesetzt.

Gruppen von Mutationen untrennbar ausführen

Sie können mit der MutationGroup-Klasse eine Gruppe von Mutationen untrennbar zusammen ausführen. Mutationen in einer MutationGroup werden garantiert in der gleichen Transaktion übertragen, aber die Transaktion wird möglicherweise wiederholt.

Mutationsgruppen funktionieren am besten, wenn sie zur Zusammenfassung von Mutationen verwendet werden, die Auswirkungen auf im Schlüsselbereich nahe beieinander gespeicherte Daten haben. Da Spanner Daten aus übergeordneten und untergeordneten Tabellen in der übergeordneten Tabelle miteinander verschachtelt, befinden sich diese Daten im Schlüsselbereich immer nahe beieinander. Wir empfehlen, die Mutationsgruppe entweder so zu strukturieren, dass sie eine auf eine übergeordnete Tabelle angewendete Mutation sowie zusätzliche auf untergeordnete Tabellen angewendete Mutationen enthält oder dass sämtliche Mutationen im Schlüsselbereich nahe beieinander liegende Daten ändern. Weitere Informationen dazu, wie Spanner Daten von über- und untergeordneten Tabellen speichert, finden Sie unter Schema und Datenmodell. Wenn Sie die Mutationsgruppen nicht anhand der empfohlenen Tabellenhierarchien strukturieren oder wenn sich die Daten, auf die zugegriffen wird, im Schlüsselbereich nicht nahe beieinander befinden, muss Spanner möglicherweise Zwei-Phasen-Commits durchführen. Das führt zu einer geringeren Leistung. Weitere Informationen finden Sie unter Kompromisse bei der Lokalität.

Wenn Sie eine MutationGroup verwenden möchten, erstellen Sie eine SpannerIO.write-Transformation und rufen Sie die SpannerIO.Write.grouped-Methode auf. Diese gibt eine Transformation zurück, die Sie dann auf eine PCollection von MutationGroup-Objekten anwenden können.

Beim Erstellen einer MutationGroup wird die erste aufgelistete Mutation zur primären Mutation. Wenn Ihre Mutationsgruppe sowohl eine übergeordnete als auch eine untergeordnete Tabelle betrifft, sollte die primäre Mutation eine Mutation der übergeordneten Tabelle sein. Andernfalls können Sie jede Mutation als primäre Mutation verwenden. Der Cloud Dataflow-Connector verwendet die primäre Mutation, um Partitionsgrenzen zu bestimmen, damit Mutationen effizient zusammengefasst werden können.

Stellen Sie sich beispielsweise vor, dass Ihre Anwendung das Verhalten der Nutzer überwacht und problematische Fälle kennzeichnet, damit diese überprüft werden. Für jeden gekennzeichneten Fall des Nutzerverhaltens möchten Sie die Tabelle Users aktualisieren, damit der Nutzer nicht mehr auf Ihre Anwendung zugreifen kann. Außerdem wollen Sie den Vorfall in der Tabelle PendingReviews aufzeichnen. Verwenden Sie eine MutationGroup, um dafür zu sorgen, dass beide Tabellen untrennbar voneinander aktualisiert werden:

GoogleSQL 

PCollection<MutationGroup> mutations =
    suspiciousUserIds.apply(
        MapElements.via(
            new SimpleFunction<>() {

              @Override
              public MutationGroup apply(String userId) {
                // Immediately block the user.
                Mutation userMutation =
                    Mutation.newUpdateBuilder("Users")
                        .set("id")
                        .to(userId)
                        .set("state")
                        .to("BLOCKED")
                        .build();
                long generatedId =
                    Hashing.sha1()
                        .newHasher()
                        .putString(userId, Charsets.UTF_8)
                        .putLong(timestamp.getSeconds())
                        .putLong(timestamp.getNanos())
                        .hash()
                        .asLong();

                // Add an entry to pending review requests.
                Mutation pendingReview =
                    Mutation.newInsertOrUpdateBuilder("PendingReviews")
                        .set("id")
                        .to(generatedId) // Must be deterministically generated.
                        .set("userId")
                        .to(userId)
                        .set("action")
                        .to("REVIEW ACCOUNT")
                        .set("note")
                        .to("Suspicious activity detected.")
                        .build();

                return MutationGroup.create(userMutation, pendingReview);
              }
            }));

mutations.apply(SpannerIO.write()
    .withInstanceId(instanceId)
    .withDatabaseId(databaseId)
    .grouped());

PostgreSQL 

PCollectionView<Dialect> dialectView =
    pipeline.apply(Create.of(Dialect.POSTGRESQL)).apply(View.asSingleton());
PCollection<MutationGroup> mutations = suspiciousUserIds
    .apply(MapElements.via(new SimpleFunction<String, MutationGroup>() {

      @Override
      public MutationGroup apply(String userId) {
        // Immediately block the user.
        Mutation userMutation = Mutation.newUpdateBuilder("Users")
            .set("id").to(userId)
            .set("state").to("BLOCKED")
            .build();
        long generatedId = Hashing.sha1().newHasher()
            .putString(userId, Charsets.UTF_8)
            .putLong(timestamp.getSeconds())
            .putLong(timestamp.getNanos())
            .hash()
            .asLong();

        // Add an entry to pending review requests.
        Mutation pendingReview = Mutation.newInsertOrUpdateBuilder("PendingReviews")
            .set("id").to(generatedId)  // Must be deterministically generated.
            .set("userId").to(userId)
            .set("action").to("REVIEW ACCOUNT")
            .set("note").to("Suspicious activity detected.")
            .build();

        return MutationGroup.create(userMutation, pendingReview);
      }
    }));

mutations.apply(SpannerIO.write()
    .withInstanceId(instanceId)
    .withDatabaseId(databaseId)
    .withDialectView(dialectView)
    .grouped());

Beim Erstellen einer Mutationsgruppe wird die erste Mutation, die als Argument angegeben ist, zur primären Mutation. In diesem Fall haben die beiden Tabellen keinen Bezug zueinander, also gibt es keine eindeutige primäre Mutation. Wir haben uns für userMutation als primäre Mutation entschieden und haben sie an die erste Stelle gesetzt. Beide Mutationen separat auszuführen wäre zwar schneller, würde aber die Untrennbarkeit nicht garantieren, weshalb die Mutationsgruppe in dieser Situation die beste Wahl darstellt.

Weitere Informationen