使用 Dataflow 匯入、匯出及修改資料

本頁面說明如何使用 Spanner 適用的 Dataflow 連接器，匯入、匯出及修改 Spanner GoogleSQL 方言資料庫和 PostgreSQL 方言資料庫中的資料。

Dataflow 是一種可轉換並充實資料的代管服務。Spanner 的 Dataflow 連接器可讓您透過 Dataflow 管道對 Spanner 讀取及寫入資料，並視需要轉換或修改資料。您也能建立在 Spanner 和其他Google Cloud 產品間轉移資料的管道。

如要有效率地將大量資料移入或移出 Spanner，建議您採用 Dataflow 連接器。此外，如果資料庫需要進行 Partitioned DML 不支援的大規模轉換作業 (例如資料表移動，以及需要 JOIN 的大量刪除作業)，建議使用這個方法。使用個別資料庫時，您也可以透過其他方式匯入及匯出資料：

  • 使用 Google Cloud 主控台將個別資料庫以 Avro 格式從 Spanner 匯出至 Cloud Storage。
  • 使用 Google Cloud 主控台將資料庫從您匯出至 Cloud Storage 的檔案重新匯入 Spanner。
  • 使用 REST API 或 Google Cloud CLI，在 Spanner 和 Cloud Storage 之間來回執行匯出或匯入工作 (同樣使用 Avro 格式)。

Spanner 的 Dataflow 連接器是 Apache Beam Java SDK 的一部分，可提供用於執行上述操作的 API。如要進一步瞭解本頁討論的部分概念，例如 PCollection 物件和轉換，請參閱 Apache Beam 程式設計指南。

將連接器新增至 Maven 專案

如要將 Google Cloud Dataflow 連接器新增至 Maven 專案，請在您的 pom.xml 檔案中新增 beam-sdks-java-io-google-cloud-platform Maven 構件做為依附元件。

舉例來說，假設您的 pom.xml 檔案將 beam.version 設定成適當的版本號碼，便會新增下列依附元件：

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

從 Spanner 讀取資料

如要從 Spanner 讀取資料，請套用 SpannerIO.read 轉換。並利用 SpannerIO.Read 類別中的方法設定讀取作業。套用上述轉換會傳回 PCollection<Struct>，其中集合內的每個元素都代表讀取作業傳回的個別資料列。您可以根據自己想要的輸出結果，選擇是否使用特定 SQL 查詢來讀取 Spanner。

套用 SpannerIO.read 轉換會執行同步讀取作業，藉此傳回一致的資料視圖。除非您另行指定，否則系統一律會在您開始讀取時建立讀取結果的快照。請參閱讀取，進一步瞭解 Spanner 可執行的各種不同讀取類型。

使用查詢讀取資料

如要從 Spanner 讀取特定一組資料，請利用 SpannerIO.Read.withQuery 方法設定轉換，藉此指定 SQL 查詢。例如：

// 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()));

不指定查詢而讀取資料

如要在不使用查詢的情況下讀取資料庫，您可以透過 SpannerIO.Read.withTable 方法指定資料表名稱，並透過 SpannerIO.Read.withColumns 方法指定要讀取的資料欄清單。例如：

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"));

如要限制讀取的資料列，可以使用 SpannerIO.Read.withKeySet 方法指定要讀取的一組主鍵。

您也可以使用指定的次要索引讀取資料表。與 readUsingIndex API 呼叫相同，索引必須包含查詢結果中顯示的所有資料。

如要這麼做，請指定上一個範例所示的資料表，並使用 SpannerIO.Read.withIndex 方法指定包含所需資料欄值的 index。索引必須儲存轉換作業需要讀取的所有資料欄。系統會隱含儲存基礎資料表的主鍵。舉例來說，如要使用索引 SongsBySongName 讀取 Songs 資料表，請使用下列程式碼：

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"));

控管交易資料的過時程度

系統保證會在一致的資料快照上執行轉換。如要控管資料的過時程度，請使用 SpannerIO.Read.withTimestampBound 方法。詳情請參閱交易。

在相同交易中讀取多份資料表

如要同時讀取多份資料表以確保資料一致性，請執行單一交易中的所有讀取作業，如要執行這項操作，請套用 createTransaction 轉換，建立 PCollectionView<Transaction> 物件，接著該物件就會建立交易。上述操作產生的視圖可透過 SpannerIO.Read.withTransaction 傳送至讀取作業。

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));

從所有可用的資料表讀取資料

您可以從 Spanner 資料庫中所有可用的資料表讀取資料。

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));

排解不支援的查詢

Dataflow 連接器僅支援 Spanner SQL 查詢，其中查詢執行計畫的第一個運算子為分散式聯集。如果您嘗試透過查詢從 Spanner 讀取資料，並發現有例外狀況說明查詢 does not have a DistributedUnion at the root，請按照「瞭解 Spanner 如何執行查詢」一節的步驟操作，利用 Google Cloud 控制台擷取查詢的執行計畫。

如果您是使用不支援的 SQL 查詢，請將該查詢簡化，讓其中的分散式聯集做為查詢執行計畫的第一個運算子。此外，請移除匯總函式、資料表聯結，以及 DISTINCT、GROUP BY 和 ORDER 運算子，因為這些運算子很可能會導致查詢無法正常運作。

建立寫入變異

請使用 Mutation 類別的 newInsertOrUpdateBuilder 方法，而非 newInsertBuilder 方法，除非 Java 管道絕對需要使用該方法。如果是 Python 管道，請使用 SpannerInsertOrUpdate，而不是 SpannerInsert。Dataflow 提供至少執行一次的強制保證，這表示系統可能會多次寫入變異。因此，只有 INSERT 變異可能會產生 com.google.cloud.spanner.SpannerException: ALREADY_EXISTS 錯誤，導致管道執行失敗。為避免發生這項錯誤，請改用 INSERT_OR_UPDATE 變異，這個方法會新增資料列，或更新已存在的資料列的資料欄值。INSERT_OR_UPDATE 異動可以套用多次。

寫入 Spanner 及轉換資料

您可以使用 SpannerIO.write 轉換執行一組輸入資料列變異，藉此透過 Dataflow 連接器將資料寫入 Spanner。為了增進效率，Dataflow 連接器會將變異批次分組。

以下範例說明如何將寫入轉換套用至變異的 PCollection：

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));

如果轉換在完成前意外停止，已套用的變異將不會復原。

自動套用變異群組

您可以使用 MutationGroup 類別，確保變異群組能一起自動套用。系統保證會經由同一項交易來提交 MutationGroup 中的變異，但也可能重新嘗試進行該交易。

如有資料是一起儲存在金鑰體系中的相近位置，您可以將影響這類資料的變數集合成變數群組，藉此發揮最佳效能。由於 Spanner 會同時將父項和子項資料表的資料交錯列於父項資料表中，這類資料在金鑰體系中的位置一向相當接近。建議您將變數群組的結構調整為以下任一形式：其中一個變異是套用至父項資料表，而其他變異則套用至子項資料表；或者所有變異都會修改金鑰體系中儲存位置相近的資料。如要進一步瞭解 Spanner 如何儲存父項和子項資料表的資料，請參閱結構定義與資料模型。如果您並未依據建議的資料表階層來整理變異群組，或您存取的資料在金鑰體系中的儲存位置並不相近，Spanner 可能會需要執行兩階段修訂，進而導致效能減慢。詳情請參閱「位置的取捨」。

如要使用 MutationGroup，請建構 SpannerIO.write 轉換並呼叫 SpannerIO.Write.grouped 方法，該方法會傳回轉換，然後您就可以將轉換套用至 MutationGroup 物件的 PCollection。

建立 MutationGroup 時，過程中列出的第一個變異會成為主要變異。如果變異群組會同時影響父項和子項資料表，主要變異應為父項資料表的變異。在其他情況下，您可以使用任何變異做為主要變異。Dataflow 連接器會利用主要變異來判定分區界線，以便有效率地將變異一同批次處理。

舉例來說，假設您的應用程式會監控行為，並標記出有問題的使用者行為供您檢查。發現帶有標記的行為時，您會希望透過更新 Users 資料表來禁止使用者存取應用程式，同時也需要將這類事件記錄在 PendingReviews 資料表。如要確保這兩份資料表皆可自動更新，請使用 MutationGroup：

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());

建立變異群組時，做為引數提供的第一個變異會成為主要變異。由於本範例中的兩份資料表互不相關，因此沒有明確的主要變異。userMutation之所以最先列出，是因為我們選取它做為主要變異。雖然分開套用兩個變異的做法比較便捷，但無法確保單元性。在此情況下，變異群組會是最理想的選擇。

後續步驟