Datastream BigQuery 移行ツールキット

このページでは、Datastream を使用して Dataflow Datastream to BigQuery テンプレートから Datastream の組み込み BigQuery レプリケーション ソリューションに移行する際のおすすめの方法について説明します。

準備

このページで説明する手順は次のとおりです。

• Docker に精通しており、インストール済みである。
• GitHub などのサービスからリポジトリのクローンを作成する方法を理解している。
• Datastream でストリームを実行する方法を理解している。
• Google Cloud CLI がインストールされている。

移行ツールキットの概要

Datastream BigQuery 移行ツールキットは、 Google Cloudが提供するオープンソース ソフトウェアです。このツールキットを使用すると Dataflow Datastream to BigQuery テンプレートから移行できます。 また、次の他のパイプラインからの移行セクションに記載されているように、他のパイプラインから移行するときにも使用できます

ツールキットを使用して BigQuery テーブルを移行するには、次の手順を行います。

1. BigQuery の宛先を使用して Datastream ストリームを作成し、開始して一時停止します。
2. 移行が必要な各 BigQuery テーブルで移行を実行します。
3. ストリームを再開します。

ツールキットは次の処理を行います。

• Datastream の Discover API を使用して、ソーステーブルのスキーマを取得します。
• 取得したスキーマに基づいて、Datastream と互換性のある BigQuery テーブルを作成します。
• 移行元の BigQuery テーブルのスキーマを取得して、必要なデータ型変換を特定します。
• 適切な列型のキャストを含め、元のテーブルの既存のすべての行を新しいテーブルにコピーします。

ツールキットの構造と使用する引数について詳しくは、ツールキットの README.md ファイルをご覧ください。

移行ツールキットを設定する

Docker を使用して移行ツールキットを実行する手順は次のとおりです。

1. リポジトリのクローンを作成し、そのディレクトリに移動します。

   git clone https://github.com/GoogleCloudPlatform/datastream-bigquery-migration-toolkit &&
   cd datastream-bigquery-migration-toolkit
   

2. イメージをビルドします。

   docker build -t migration-service .
   

3. Google Cloud CLI の認証情報を使用して認証します。

   docker run -ti \
   --name gcloud-config migration-service gcloud auth application-default login
   

4. Google Cloud プロジェクト プロパティを設定します。

   docker run -ti --volumes-from gcloud-config migration-service \
   gcloud config set project PROJECT_ID
   

   PROJECT_ID は、 Google Cloudプロジェクトの ID に置き換えます。

Dataflow から Datastream の組み込みソリューションに移行する

1. BigQuery を宛先とするストリームを作成する

   注: ソースを構成するときは、[履歴データのバックフィル モードを選択] で [手動バックフィル] を選択してください。

2. ストリームを開始し、すぐに一時停止します。これにより、Datastream は移行の開始前に変更をキャプチャする位置をキャプチャできます。

   注: 開始から一時停止までの間に、Datastream が一部のイベントを処理し、BigQuery に 1 つ以上のテーブルを作成する場合があります。このような場合は、移行を続行する前に BigQuery でテーブルを手動で削除する必要があります。削除しないと、移行は失敗します。

3. Datastream と Dataflow パイプラインをドレインします。

   1. 既存の Cloud Storage 転送先ストリームを一時停止します。
   2. ストリームの合計レイテンシ指標を確認し、少なくとも現在のレイテンシと同じ時間待機して、処理中のイベントが宛先に書き込まれるようにします。
   3. Dataflow ジョブをドレインします。

4. 移行を実行します。

   1. 移行を dry_run モードで実行します。dry_run モードでは、CREATE TABLE DDL ステートメントとデータをコピーする SQL ステートメントを生成できます。これらのステートメントは実行されません。

      docker run -v output:/output -ti --volumes-from gcloud-config \
      migration-service python3 ./migration/migrate_table.py dry_run \
      --project-id PROJECT_ID \
      --stream-id STREAM_ID \
      --datastream-region STREAM_REGION \
      --source-schema-name SOURCE_SCHEMA_NAME \
      --source-table-name SOURCE_TABLE_NAME \
      --bigquery-source-dataset-name BIGQUERY_SOURCE_DATASET_NAME \
      --bigquery-source-table-name BIGQUERY_SOURCE_TABLE_NAME
      

      次のように置き換えます。

      • PROJECT_ID: Google Cloudプロジェクトの固有識別子。
      • STREAM_ID: BigQuery 宛先ストリームの一意の識別子。
      • STREAM_REGION: ストリームのロケーション（例: us-central1）。
      • SOURCE_SCHEMA_NAME: ソーススキーマの名前。
      • SOURCE_TABLE_NAME: ソーステーブルの名前。
      • BIGQUERY_SOURCE_DATASET_NAME: 既存の BigQuery データセットの名前。
      • BIGQUERY_SOURCE_TABLE_NAME: 既存の BigQuery テーブルの名前。

   2. output/create_target_table と output/copy_rows の下の .sql ファイルを調べます。 Google Cloud プロジェクトで実行される SQL コマンドは次のとおりです。

      docker run -v output:/output -ti migration-service find output/create_target_table \
      -type f -print -exec cat {} \;
      

      docker run -v output:/output -ti migration-service find output/copy_rows \
      -type f -print -exec cat {} \;
      

   3. SQL コマンドを実行するには、移行を full モードで実行します。full モードでは、BigQuery にテーブルを作成し、既存の BigQuery テーブルからすべての行をコピーできます。

      docker run -v output:/output -ti --volumes-from gcloud-config \
      migration-service python3 ./migration/migrate_table.py full \
      --project-id PROJECT_ID \
      --stream-id STREAM_ID \
      --datastream-region STREAM_REGION \
      --source-schema-name SOURCE_SCHEMA_NAME \
      --source-table-name SOURCE_TABLE_NAME \
      --bigquery-source-dataset-name BIGQUERY_SOURCE_DATASET_NAME \
      --bigquery-source-table-name BIGQUERY_SOURCE_TABLE_NAME
      

5. 一時停止したストリームを再開します。

6. Google Cloud ログ エクスプローラを開き、次のクエリを使用して Datastream ログを探します。

   resource.type="datastream.googleapis.com/Stream"
   resource.labels.stream_id=STREAM_ID
   

   次のログを探します。ここで、%d は数値です。

   Completed writing %d records into..
   

   このログは、新しいストリームが BigQuery にデータが正常に読み込まれたことを示しています。読み込む CDC データがある場合にのみ表示されます。

他のパイプラインから移行する

このツールキットを使用すると、他のパイプラインから Datastream の組み込み BigQuery ソリューションに移行することもできます。ツールキットでは、dry_run モードを使用して、ソース データベース スキーマに基づいて Datastream 互換のある BigQuery テーブルの CREATE TABLE DDL を生成できます。

  docker run -v output:/output -ti --volumes-from gcloud-config \
  migration-service python3 ./migration/migrate_table.py dry_run \
  --project-id PROJECT_ID \
  --stream-id STREAM_ID \
  --datastream-region STREAM_REGION \
  --source-schema-name SOURCE_SCHEMA_NAME \
  --source-table-name SOURCE_TABLE_NAME \
  --bigquery-source-dataset-name BIGQUERY_SOURCE_DATASET_NAME \
  --bigquery-source-table-name BIGQUERY_SOURCE_TABLE_NAME

BigQuery テーブルのスキーマは異なる可能性があるため、行をコピーするための汎用的な SQL ステートメントを提供することは困難です。ソーステーブルの DDL ディレクトリ（output/source_table_ddl）と宛先テーブルの DDL ディレクトリ（output/create_target_table）を使用すると、参照元の列の適切なキャストを用いて、SQL ステートメントを作成します。

使用できる SQL ステートメントの形式の例を次に示します。

  INSERT INTO DESTINATION_TABLE (DESTINATION_COLUMN1, DESTINATION_COLUMN2...)
  SELECT SOURCE_COLUMN1, SOURCE_COLUMN2
  FROM SOURCE_TABLE;

以下を置き換えます。

• DESTINATION_TABLE: BigQuery の出力先テーブルの名前。
• DESTINATION_COLUMN: 宛先テーブルの列の名前。
• SOURCE_COLUMN: ソーステーブルの列の名前。
• SOURCE_TABLE: ソーステーブルの名前。

制限事項

ツールキットには次の制限があります。

• 既存の BigQuery テーブルと新しい BigQuery テーブルの列名は一致している必要があります（メタデータ列は無視されます）。既存のテーブルの列名を変更する（たとえば、接頭辞を追加するか、大文字小文字の変更など） Dataflow ユーザー定義関数（UDF）を適用をした場合、移行が失敗します。
• クロスリージョン移行とクロス プロジェクト移行はサポートされていません。
• 移行はテーブル単位で行われます。
• Oracle と MySQL のソースのみがサポートされています。