Cloud Storage Text to Spanner テンプレート

Cloud Storage Text to Spanner テンプレートは、Storage から CSV テキストファイルを読み取り、Spanner データベースにインポートするバッチパイプラインです。

パイプラインの要件

ターゲットの Spanner データベースとテーブルが存在している必要があります。
Cloud Storage バケットの読み取り権限と、対象の Spanner データベースに対する書き込み権限が必要です。
CSV ファイルを含む入力 Cloud Storage パスが存在している必要があります。
CSV ファイルの JSON 記述を含むインポートマニフェストファイルを作成し、そのマニフェストを Cloud Storage に保存する必要があります。
ターゲットの Spanner データベースにすでにスキーマがある場合、マニフェストファイルで指定された列は、ターゲットデータベースのスキーマ内の対応する列と同じデータ型である必要があります。

ASCII または UTF-8 でエンコードされたマニフェストファイルは、次の形式に一致する必要があります。

マニフェストの形式と例

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

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"}
      ]
    }
  ]
}

インポートするテキストファイルは、ASCII または UTF-8 エンコードの CSV 形式である必要があります。UTF-8 エンコードファイルではバイトオーダーマーク（BOM）を使用しないことをおすすめします。

データは次のタイプのいずれかに一致する必要があります。

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

テンプレートのパラメータ

必須パラメータ

instanceId : Spanner データベースのインスタンス ID。
databaseId : Spanner データベースのデータベース ID。
importManifest : マニフェストファイルのインポート時に使用する Cloud Storage のパス（例: gs://your-bucket/your-folder/your-manifest.json）。

オプションパラメータ

spannerHost: テンプレートで呼び出す Cloud Spanner エンドポイント。テスト専用です（例: https://batch-spanner.googleapis.com）。デフォルトは https://batch-spanner.googleapis.com です。
columnDelimiter : ソースファイルが使用する列区切り文字。デフォルト値は「,」です（例: ,）。
fieldQualifier: columnDelimiter を含むソースファイルで値を囲む必要がある文字。デフォルト値は「"」です。
trailingDelimiter : ソースファイルの行の末尾に区切り文字があるかどうかを指定します（つまり、columnDelimiter 文字が各行の最後の列の値の後に表示されるかどうか）。デフォルト値は true です。
escape : ソースファイルが使用するエスケープ文字。デフォルトでは、このパラメータは設定されておらず、テンプレートでエスケープ文字は使用されません。
nullString : NULL 値を表す文字列。デフォルトでは、このパラメータは設定されておらず、テンプレートで null 文字列は使用されません。
dateFormat : 日付列を解析するために使用される形式。デフォルトでは、パイプラインは日付列を yyyy-M-d[' 00:00:00'] として解析します。たとえば 2019-01-31 または 2019-1-1 00:00:00 となります。日付形式が異なる場合は、java.time.format.DateTimeFormatter（https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/time/format/DateTimeFormatter.html）のパターンを使用して形式を指定します。
timestampFormat : timestamp 列を解析するために使用される形式。タイムスタンプが Long 整数の場合は、Unix エポック時間として解析されます。それ以外の場合は、java.time.format.DateTimeFormatter.ISO_INSTANT（https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/time/format/DateTimeFormatter.html#ISO_INSTANT）形式を使用して、文字列として解析されます。その他の場合では、独自のパターン文字列を指定します。たとえば、"Jan 21 1998 01:02:03.456+08:00" という形式のタイムスタンプには、「MMM dd yyyy HH:mm:ss.SSSVV」を使用します。
spannerProjectId : Spanner データベースを含む Google Cloud プロジェクトの ID。設定されていない場合は、デフォルトの Google Cloud プロジェクトのプロジェクト ID が使用されます。
spannerPriority : Spanner 呼び出しのリクエストの優先度。指定できる値は、HIGH、MEDIUM、LOW です。デフォルト値は MEDIUM です。
handleNewLine : true の場合、入力データに改行文字を含めることができます。それ以外の場合は、改行文字が原因でエラーが発生します。デフォルト値は false です。改行処理を有効にすると、パフォーマンスが低下する可能性があります。
invalidOutputPath : インポートできない行を書き込むときに使用する Cloud Storage パス（例: gs://your-bucket/your-path）。デフォルトは空です。

日付形式やタイムスタンプ形式をカスタマイズする必要がある場合は、有効な java.time.format.DateTimeFormatter パターンであることを確認してください。次の表に、カスタマイズされた date 列と timestamp 列の形式の例を示します。

型	入力値	フォーマット	備考
`DATE`	2011-3-31		デフォルトでは、テンプレートはこの形式を解析できます。`dateFormat` パラメータを指定する必要はありません。
`DATE`	2011-3-31 00:00:00		デフォルトでは、テンプレートはこの形式を解析できます。形式を指定する必要はありません。必要に応じて `yyyy-M-d' 00:00:00'` が使用できます。
`DATE`	2018 年 4 月 1 日	dd MMM, yy
`DATE`	西暦 2019 年 4 月 3 日水曜日	EEEE, LLLL d, yyyy G
`TIMESTAMP`	2019-01-02T11:22:33Z 2019-01-02T11:22:33.123Z 2019-01-02T11:22:33.12356789Z		デフォルトの形式 `ISO_INSTANT` はこのタイプのタイムスタンプを解析できます。`timestampFormat` パラメータを指定する必要はありません。
`TIMESTAMP`	1568402363		デフォルトでは、テンプレートはこのタイプのタイムスタンプを解析し、Unix エポック時間として扱います。
`TIMESTAMP`	2008 年 6 月 3 日（火曜日）11:05:30 GMT	EEE, d MMM yyyy HH:mm:ss VV
`TIMESTAMP`	2018 年 12 月 31 日 110530.123 太平洋標準時	yyyy/MM/dd HHmmss.SSSz
`TIMESTAMP`	2019-01-02T11:22:33Z または 2019-01-02T11:22:33.123Z	yyyy-MM-dd'T'HH:mm:ss [.SSS]VV	入力列が 2019-01-02T11:22:33Z と 2019-01-02T11:22:33.123Z の場合、この形式のタイムスタンプはデフォルトの形式で解析できます。独自のフォーマットパラメータを指定する必要はありません。`yyyy-MM-dd'T'HH:mm:ss[.SSS]VV` を使用すると、両方のケースに対応できます。接尾辞「Z」は文字リテラルではなくタイムゾーン ID として解析する必要があるため、`yyyy-MM-dd'T'HH:mm:ss[.SSS]'Z'` は使用できません。内部的には、タイムスタンプ列は `java.time.Instant` に変換されます。そのため、UTC で指定するか、タイムゾーン情報を設定する必要があります。2019-01-02 11:22:33 のようなローカル日時は、有効な `java.time.Instant` として解析されません。

テンプレートを実行する

コンソール

Dataflow の [テンプレートからジョブを作成] ページに移動します。

[テンプレートからジョブを作成] に移動

[ジョブ名] フィールドに、固有のジョブ名を入力します。
（省略可）[リージョンエンドポイント] で、プルダウンメニューから値を選択します。デフォルトのリージョンは us-central1 です。
Dataflow ジョブを実行できるリージョンのリストについては、Dataflow のロケーションをご覧ください。
[Dataflow テンプレート] プルダウンメニューから、[ the Text Files on Cloud Storage to Cloud Spanner template] を選択します。
表示されたパラメータフィールドに、パラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

シェルまたはターミナルで、テンプレートを実行します。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/GCS_Text_to_Cloud_Spanner \
    --region REGION_NAME \
    --parameters \
instanceId=INSTANCE_ID,\
databaseId=DATABASE_ID,\
importManifest=GCS_PATH_TO_IMPORT_MANIFEST

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

JOB_NAME: 一意の任意のジョブ名
VERSION: 使用するテンプレートのバージョン
使用できる値は次のとおりです。
- latest: 最新バージョンのテンプレートを使用します。このテンプレートは、バケット内で日付のない親フォルダ（gs://dataflow-templates-REGION_NAME/latest/）にあります。
- バージョン名（例: 2023-09-12-00_RC00）。特定のバージョンのテンプレートを使用します。このテンプレートは、バケット内で対応する日付の親フォルダ（gs://dataflow-templates-REGION_NAME/）にあります。
注: 最新のテンプレートでは、互換性のない変更が行われている場合があります。こうした互換性のない変更が本番環境のワークフローに影響しないように、本番環境では最新の日付付き親フォルダに保存されているテンプレートを使用する必要があります。
REGION_NAME: Dataflow ジョブをデプロイするリージョン（例: us-central1）
INSTANCE_ID: Spanner インスタンス ID
DATABASE_ID: Spanner データベース ID
GCS_PATH_TO_IMPORT_MANIFEST: インポートマニフェストファイルへの Cloud Storage パス

API

REST API を使用してテンプレートを実行するには、HTTP POST リクエストを送信します。API とその認証スコープの詳細については、projects.templates.launch をご覧ください。

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates-LOCATION/VERSION/GCS_Text_to_Cloud_Spanner
{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "importManifest": "GCS_PATH_TO_IMPORT_MANIFEST"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}