CSV 形式でのデータのインポートとエクスポート

このページでは、Spanner から CSV ファイルにデータをエクスポートする方法、または CSV ファイルから Spanner データベースにデータをインポートする方法について説明します。

以前に Cloud Storage の Avro ファイルにエクスポートした Spanner データベースをインポートする場合は、Spanner Avro ファイルをインポートするをご覧ください。
Spanner 以外のデータベースから Avro ファイルをインポートする場合は、Spanner 以外のデータベースからのデータインポートをご覧ください。

このプロセスでは Dataflow を使用します。Spanner から Cloud Storage バケットにデータをエクスポートできます。または、JSON マニフェストファイルと一連の CSV ファイルを含む Cloud Storage バケットから Spanner にデータをインポートすることもできます。

始める前に

Spanner データベースをインポートするには、まず Spanner、Cloud Storage、Compute Engine、Dataflow API を有効にする必要があります。

API を有効にする

また、十分な割り当てと必須の IAM 権限も必要です。

割り当て要件

インポートジョブまたはエクスポートジョブの割り当て要件は次のとおりです。

Spanner: インポートするデータの量をサポートできるだけのコンピューティング容量が必要です。データベースをインポートまたはエクスポートするために追加のコンピューティング容量は必要ありませんが、ジョブが妥当な時間内に終了するようにコンピューティング容量を追加する必要がある場合があります。詳細については、ジョブを最適化するをご覧ください。
Cloud Storage: インポートするには、以前にエクスポートしたファイルが格納されているバケットが必要です。エクスポートするには、エクスポートされたファイル用のバケットを作成する必要があります（まだない場合）。これは、Google Cloud コンソールで Cloud Storage ページを使用して、または Spanner ページを使用してエクスポートを作成するときに行うことができます。バケットのサイズを設定する必要はありません。
Dataflow: インポートジョブまたはエクスポートジョブは、他の Dataflow ジョブと同じ CPU、ディスク使用量、IP アドレスの Compute Engine の割り当てに従います。
Compute Engine: インポートジョブまたはエクスポートジョブを実行する前に、Dataflow によって使用される Compute Engine の初期割り当てを設定する必要があります。これらの割り当ては、Dataflow でジョブに使用できる最大リソース数を表します。推奨の開始値は次のとおりです。
- CPU: 200
- 使用中の IP アドレス: 200
- 標準永続ディスク: 50 TB
通常、他の調整は必要ありません。Dataflow では自動スケーリングが提供されているため、インポートまたはエクスポート中に実際に使用したリソースに対してのみ料金を支払います。ジョブでより多くのリソースが使用される可能性がある場合、Dataflow UI に警告アイコンが表示されます。警告アイコンが表示されてもジョブは完了します。

IAM 要件

データベースをインポートまたはエクスポートするには、インポートジョブまたはエクスポートジョブに関連するすべてのサービスを使用するための十分な権限を持つ IAM のロールも必要です。ロールと権限の付与については、IAM ロールを適用するをご覧ください。

データベースをインポートまたはエクスポートするには、次の役割が必要です。

Google Cloud プロジェクトレベル:
- Spanner 閲覧者（roles/spanner.viewer）
- Dataflow 管理者（roles/dataflow.admin）
- Storage 管理者（roles/storage.admin）
Spanner データベースまたはインスタンスレベル、あるいは Google Cloud プロジェクトレベル:
- Spanner データベースリーダー（roles/spanner.databaseReader）
- Spanner データベース管理者（roles/spanner.databaseAdmin）

Spanner データを CSV ファイルにエクスポートする

Spanner から Cloud Storage の CSV ファイルにデータをエクスポートするには、Google Cloud CLI を使用して Spanner to Cloud Storage Text テンプレートでジョブを実行する手順を行います。

このドキュメントのジョブの表示またはトラブルシューティング、遅いジョブの最適化、ジョブのパフォーマンスに影響する要素の情報を参照することもできます。

CSV ファイルから Spanner にデータをインポートする

CSV ファイルからデータをインポートするプロセスには、次のステップが含まれます。

データを CSV ファイルにエクスポートし、それらのファイルを Cloud Storage に保存します。ヘッダー行は含めないでください。
JSON マニフェストファイルを作成し、CSV ファイルとともに保存します。
Spanner データベースに空のターゲットテーブルを作成するか、CSV ファイルの列のデータ型が既存のテーブルの対応する列と一致することを確認します。
インポートジョブを実行します。

手順 1: Spanner 以外のデータベースから CSV ファイルにデータをエクスポートする

インポートプロセスは、Cloud Storage バケット内の CSV ファイルからデータを取り込みます。任意のソースからデータを CSV 形式でエクスポートできます。

データをエクスポートするときは、次の点に留意してください。

インポートするテキストファイルは CSV 形式である必要があります。
データは次のタイプのいずれかに一致する必要があります。

GoogleSQL

BOOL
INT64
FLOAT64
NUMERIC
STRING
DATE
TIMESTAMP
BYTES
JSON

PostgreSQL

boolean
bigint
double precision
numeric
character varying, text
date
timestamp with time zone
bytea

CSV ファイルをエクスポートする際に、メタデータを含めたり生成したりする必要はありません。
ファイルについて特定の命名規則に従う必要はありません。

ファイルを Cloud Storage に直接エクスポートしない場合は、Cloud Storage バケットに CSV ファイルをアップロードする必要があります。

ステップ 2: JSON マニフェストファイルを作成する

インポートするファイルの JSON 記述を含むマニフェストファイルも作成し、CSV ファイルが保存されている Cloud Storage バケットと同じバケットに配置する必要があります。このマニフェストファイルには、各テーブルの名前とデータファイルの場所を示す tables 配列を格納します。このファイルでは、受信側のデータベース言語も指定します。言語を省略すると、デフォルトで GoogleSQL が使用されます。

次のメッセージタイプに対応するマニフェストファイルの形式は、プロトコルバッファで参照できます。

message ImportManifest {
  // The per-table import manifest.
  message TableManifest {
    // Required. The name of the destination table.
    string table_name = 1;
    // Required. The CSV files to import. This value can be either a filepath or a glob pattern.
    repeated string file_patterns = 2;
    // The schema for a table column.
    message Column {
      // Required for each Column that you specify. The name of the column in the
      // destination table.
      string column_name = 1;
      // Required for each Column that you specify. The type of the column.
      string type_name = 2;
    }
    // Optional. The schema for the table columns.
    repeated Column columns = 3;
  }
  // Required. The TableManifest of the tables to be imported.
  repeated TableManifest tables = 1;

  enum ProtoDialect {
    GOOGLE_STANDARD_SQL = 0;
    POSTGRESQL = 1;
  }
  // Optional. The dialect of the receiving database. Defaults to GOOGLE_STANDARD_SQL.
  ProtoDialect dialect = 2;
}

次の例は、Albums と Singers というテーブルを GoogleSQL 言語データベースにインポートするマニフェストファイルを示しています。Albums テーブルは、ジョブがデータベースから取得する列スキーマを使用し、Singers テーブルはマニフェストファイルが指定するスキーマを使用します。

{
  "tables": [
    {
      "table_name": "Albums",
      "file_patterns": [
        "gs://bucket1/Albums_1.csv",
        "gs://bucket1/Albums_2.csv"
      ]
    },
    {
      "table_name": "Singers",
      "file_patterns": [
        "gs://bucket1/Singers*.csv"
      ],
      "columns": [
        {"column_name": "SingerId", "type_name": "INT64"},
        {"column_name": "FirstName", "type_name": "STRING"},
        {"column_name": "LastName", "type_name": "STRING"}
      ]
    }
  ]
}

ステップ 3: Spanner データベースのテーブルを作成する

インポートを実行する前に、Spanner データベースにターゲットテーブルを作成する必要があります。ターゲットの Spanner テーブルにすでにスキーマがある場合、マニフェストファイルで指定された列は、ターゲットテーブルのスキーマ内の対応する列と同じデータ型である必要があります。

セカンダリインデックス、外部キー、変更ストリームは、最初にテーブルを作成するときではなく、Spanner にデータをインポートした後に作成することをおすすめします。テーブルにこれらの構造がすでに含まれている場合は、それらを削除して、データをインポートした後に再作成することをおすすめします。

ステップ 4: gcloud を使用して Dataflow インポートジョブを実行する

インポートジョブを開始するには、Google Cloud CLI を使用して、Cloud Storage Text to Spanner テンプレートでジョブを実行する手順を行います。

インポートジョブを開始したら、Google Cloud コンソールでジョブの詳細を確認できます。

インポートジョブが完了したら、必要なセカンダリインデックス、外部キー、変更ストリームを追加します。

インポートジョブのリージョンの選択

Cloud Storage バケットのロケーションに基づいて別のリージョンを選択することもできます。送信データ転送料金を回避するには、Cloud Storage バケットのロケーションと一致するリージョンを選択します。

Cloud Storage バケットのロケーションがリージョンである場合、次の前提で、インポートジョブに同じリージョンを選択することで、無料のネットワーク使用を利用できます。
Cloud Storage バケットのロケーションがデュアルリージョンである場合、いずれかのリージョンが使用可能であると仮定すると、構成する 2 つのリージョンのいずれかを選択して、無料のネットワーク使用を利用できます。
インポートジョブが同じ場所に配置できない場合、または Cloud Storage バケットのロケーションがマルチリージョンである場合、送信データ転送料金が適用されます。Cloud Storage のデータ転送の料金を参照して、データ転送の料金が最低になるリージョンを選択してください。

Dataflow UI でジョブを表示またはトラブルシューティングする

インポートジョブまたはエクスポートジョブの開始後、Google Cloud コンソールの Dataflow セクションで、ジョブの詳細（ログなど）を表示できます。

Dataflow ジョブの詳細を表示する

現在実行中のジョブを含む過去 1 週間以内のインポート / エクスポートジョブの詳細を表示するには、次のコマンドを実行します。

データベースの [データベースの詳細] ページに移動します。
左側のパネルにある [インポート / エクスポート] をクリックします。データベースの [インポート / エクスポート] ページに、最近のジョブのリストが表示されます。
データベースの [インポート / エクスポート] ページで、[Dataflow ジョブ名] 列のジョブ名をクリックします。

Google Cloud コンソールに Dataflow ジョブの詳細が表示されます。

1 週間以上前に実行したジョブを表示するには:

Google Cloud コンソールの Dataflow ジョブページに移動します。

ジョブページに移動
リスト内でジョブを見つけ、その名前をクリックします。

Google Cloud コンソールに Dataflow ジョブの詳細が表示されます。

ジョブの Dataflow ログを表示する

Dataflow ジョブのログを表示するには、上記の説明に従ってジョブの詳細ページに移動し、ジョブ名の右側にある [ログ] をクリックします。

ジョブが失敗した場合は、ログでエラーを探します。エラーがある場合、エラー数が [ログ] の横に表示されます。

[ログ] ボタンの横のエラー数の例

ジョブエラーを表示するには:

[ログ] の横のエラー数をクリックします。

Google Cloud コンソールにジョブのログが表示されます。エラーを表示するには、スクロールが必要な場合があります。
エラーアイコンが表示されているエントリを見つけます。
個別のログエントリをクリックして、その内容を展開します。

Dataflow ジョブのトラブルシューティングの詳細については、パイプラインをトラブルシューティングするをご覧ください。

失敗したインポートジョブまたはエクスポートジョブのトラブルシューティングを実施する

ジョブログに次のエラーが表示された場合、

com.google.cloud.spanner.SpannerException: NOT_FOUND: Session not found

--or--

com.google.cloud.spanner.SpannerException: DEADLINE_EXCEEDED: Deadline expired before operation could complete.

Google Cloud コンソールで Spanner データベースの [モニタリング] タブで、99% の読み取り / 書き込みレイテンシを確認します。高い値（数秒）が表示されている場合は、インスタンスが過負荷状態であるため、読み取り / 書き込みがタイムアウトになって失敗します。

レイテンシが高くなる原因の一つは、過剰な数のワーカーを使用して Dataflow ジョブが走行しているため、Spanner インスタンスに過剰な負荷がかかることです。

Dataflow ワーカーの数の上限を指定するには:

Dataflow コンソールを使用している場合、[最大ワーカー数] パラメータは、[テンプレートからジョブを作成] ページの [オプションのパラメータ] セクションに配置されます。

gcloud を使用している場合は、max-workers 引数を指定します。例:

gcloud dataflow jobs run my-import-job \
--gcs-location='gs://dataflow-templates/latest/GCS_Text_to_Cloud_Spanner' \
--region=us-central1 \
--parameters='instanceId=test-instance,databaseId=example-db,inputDir=gs://my-gcs-bucket' \
--max-workers=10

実行速度が遅いインポートジョブまたはエクスポートジョブを最適化する

初期設定の提案に従っている場合は、通常、他の調整は必要ありません。ジョブの実行速度が遅い場合は、その他の最適化を試すことができます。

ジョブとデータのロケーションの最適化: Spanner インスタンスと Cloud Storage バケットが配置されている同じリージョン内で Dataflow ジョブを実行します。
十分な Dataflow リソースの確保: 関連する Compute Engine の割り当てによって Dataflow ジョブのリソースが制限されている場合、Google Cloud Console の、ジョブの Dataflow ページに警告アイコンとログメッセージが表示されます。

この場合、CPU、使用中の IP アドレス、標準永続ディスクの割り当てを増やすと、ジョブの実行時間が短くなる可能性がありますが、Compute Engine の追加料金が発生する場合があります。
Spanner の CPU 使用率の確認: インスタンスの CPU 使用率が 65% を超えている場合は、そのインスタンスのコンピューティング容量を増やすことができます。容量を追加すると Spanner のリソースが増加し、ジョブの実行速度は速くなりますが、Spanner の追加料金が発生します。

インポートジョブまたはエクスポートジョブのパフォーマンスに影響する要素

インポートジョブまたはエクスポートジョブの完了にかかる時間には、いくつかの要素が影響します。

Spanner データベースのサイズ: 処理するデータ量が増加すると、必要となる時間とリソースも多くなります。
Spanner データベーススキーマ。次のものが含まれます。
- テーブルの数
- 行のサイズ
- セカンダリインデックス数
- 外部キーの数
- 変更ストリームの数

データのロケーション: データは、Dataflow により Spanner と Cloud Storage 間で転送されます。3 つのコンポーネントがすべて同じリージョン内にあることが理想的です。コンポーネントが同じリージョン内にない場合は、リージョン間のデータの移動によってジョブは遅くなります。
Dataflow ワーカーの数: パフォーマンスの向上には、最適な Dataflow ワーカーが必要です。自動スケーリングを使用することにより、Dataflow では、処理する必要がある作業量に応じてジョブのワーカー数が選択されます。ただし、CPU、使用中の IP アドレス、標準永続ディスクの割り当てにより、ワーカー数には上限があります。割り当ての上限に達すると、Dataflow UI に警告アイコンが表示されます。この状況では、進捗は遅くなりますがジョブは完了します。インポートするデータの量が多い場合は、自動スケーリングによって Spanner が過負荷状態になり、エラーが発生する可能性があります。
Spanner に対する既存の負荷: インポートジョブを実行すると、Spanner インスタンスに対する CPU 負荷が大幅に上昇します。エクスポートジョブの場合は通常、Spanner インスタンスへの負荷は軽くなります。インスタンスに既存の負荷がかなりある場合、このジョブの実行速度はさらに遅くなります。
Spanner のコンピューティング容量: インスタンスの CPU 使用率が 65% を超えると、ジョブの実行速度はさらに低下します。

インポートパフォーマンスの向上のためにワーカーを調整する

Spanner のインポートジョブを開始する際は、良好なパフォーマンスを実現するために、Dataflow ワーカーを最適な値に設定する必要があります。過剰な数のワーカーが存在すると Spanner が過負荷状態になり、ワーカー数が過剰に少ない場合はインポートのパフォーマンスが低下します。

ワーカーの最大数はデータサイズに大きく依存しますが、Spanner の CPU 使用率の合計は 70～90% が理想的です。これにより、Spanner の効率性とエラーのないジョブ完了のバランスが良好になります。

大部分のスキーマ / シナリオでその使用率の目標を達成するには、ワーカー vCPU の最大数を Spanner ノードの数の 4〜6 倍にすることをおすすめします。

たとえば、n1-standard-2 ワーカーを使用する 10 ノードの Spanner インスタンスの場合、最大ワーカー数を 25 に設定し、50 個の vCPU を割り当てます。