Cloud Storage の転送

BigQuery Data Transfer Service for Cloud Storage を使用すると、Cloud Storage から BigQuery への定期的なデータ読み込みをスケジュールできます。

始める前に

Cloud Storage の転送を作成する前に、以下を行います。

BigQuery Data Transfer Service の有効化に必要なすべての操作が完了していることを確認します。
Cloud Storage URI を取得します。
データを保存する BigQuery データセットを作成します。
転送用に宛先テーブルを作成し、スキーマ定義を指定します。

制限事項

Cloud Storage から BigQuery への繰り返しの転送には、次の制限があります。

転送のワイルドカードまたはランタイムパラメータで定義したパターンに一致するファイルはすべて、宛先テーブルに対して定義したスキーマと同じスキーマを共有している必要があり、そうでない場合は転送が失敗します。転送後にテーブルスキーマを変更して再度転送を実行した場合も、転送が失敗します。
Cloud Storage オブジェクトはバージョニングが可能ですが、アーカイブ済みの Cloud Storage オブジェクトは BigQuery に転送できないことに注意してください。転送されるオブジェクトは、ライブデータである必要があります。
Cloud Storage から BigQuery へ個別にデータを読み込む場合とは異なり、継続的な転送では転送を設定する前に宛先テーブルとスキーマを作成する必要があります。BigQuery では、繰り返しのデータ転送プロセスの一環としてテーブルを作成することはできません。
Cloud Storage からの転送は、常に WRITE_APPEND 設定によってトリガーされ、データが宛先テーブルに追加されます。詳細については、load ジョブ構成の configuration.load.writeDisposition をご覧ください。
BigQuery Data Transfer Service では、転送の最中に Cloud Storage のファイルがアクセスされた場合、すべてのファイルが転送されること、あるいはファイルが 1 回のみ転送されることは保証されません。
データセットのロケーションが US 以外の値に設定されている場合は、リージョンまたはマルチリージョンの Cloud Storage バケットがデータセットと同じリージョンに存在する必要があります。
BigQuery Data Transfer Service では外部データソースのデータ整合性は保証されません。クエリの実行中に基になるデータを変更すると、予期しない動作が発生する可能性があります。
Cloud Storage のソースデータの形式によっては、追加の制限が適用される場合があります。詳細については、次をご覧ください。
Cloud Storage バケットは、BigQuery の宛先データセットのリージョンまたはマルチリージョンと互換性のあるリージョンまたはマルチリージョンに存在する必要があります。これはコロケーションと呼ばれます。詳細については、Cloud Storage 転送のデータのロケーションをご覧ください。

最小間隔

Cloud Storage 内のソースファイルが転送の対象になるには、そのファイルが作成後 1 時間以上経過している必要があります。
定期的な転送の最小間隔は 1 時間です。デフォルトの定期的な転送間隔は 24 時間です。

必要な権限

BigQuery にデータを読み込む場合は、新規または既存の BigQuery のテーブルやパーティションにデータを読み込むための権限が必要です。Cloud Storage からデータを読み込む場合は、データが格納されているバケットへのアクセス権も必要です。次の必要な権限があることを確認します。

BigQuery: スケジュールされた転送を作成するには、bigquery.transfers.update 権限が必要です。bigquery.transfers.update 権限は、IAM 事前定義ロール bigquery.admin に含まれています。BigQuery Data Transfer Service での IAM ロールの詳細については、アクセス制御のリファレンスをご覧ください。
Cloud Storage: 個々のバケット以上のレベルで storage.objects.get 権限が必要です。URI のワイルドカードを使用する場合は storage.objects.list 権限も必要です。転送が完了するたびにソースファイルを削除する場合は、storage.objects.delete 権限も必要です。これらすべての権限は、事前定義の IAM のロール storage.objectAdmin に含まれています。

Cloud Storage の転送の設定

BigQuery Data Transfer Service で Cloud Storage 転送を作成するには:

Console

Cloud Console の BigQuery ページに移動します。

[BigQuery] ページに移動
[転送] をクリックします。
[作成] をクリックします。
[転送の作成] ページで、次の操作を行います。
- [ソースタイプ] セクションで、[ソース] として [Cloud Storage] を選択します。
- [転送構成名] セクションの [表示名] に、転送名（例: My Transfer）を入力します。転送名には、後で修正が必要になった場合に簡単に識別できる任意の名前を使用できます。
- [スケジュールオプション] セクションで、[スケジュール] をデフォルト値（[すぐに開始可能]）のままにするか、[設定した時間に開始] をクリックします。
  - [繰り返しの頻度] で、転送を実行する頻度のオプションを選択します。最小間隔は 1 時間です。
    - 毎日（デフォルト）
    - 毎週
    - 毎月
    - カスタム
    - オンデマンド
  - [開始日と実行時間] に、転送を開始する日付と時刻を入力します。[すぐに開始可能] を選択した場合、このオプションは無効になります。
- [転送先の設定] セクションの [宛先データセット] で、データを保存するために作成したデータセットを選択します。
- [データソースの詳細] セクションで、次の操作を行います。
  - [宛先テーブル] に宛先テーブルの名前を入力します。宛先テーブルは、テーブルの命名規則に従う必要があります。宛先テーブル名でもパラメータがサポートされています。
  - [Cloud Storage URI] に Cloud Storage URI を入力します。ワイルドカードとパラメータがサポートされています。
  - [書き込み設定] の場合、次から選択します。
    - APPEND で、既存の宛先テーブルに新しいデータを追加する。
    - MIRROR で、宛先テーブル内のデータを更新し、変更されたデータをソースに反映する。MIRROR は、宛先テーブル内のデータの新しいコピーを上書きします。
    注: [Cloud Storage URI] フィールドに入力した内容がどのファイルにも一致しない場合、宛先テーブルのデータを更新（上書き）しません。
  - 転送が完了するたびにソースファイルを削除する場合は、[Delete source files after transfer] をオンにします。削除ジョブはベストエフォートベースです。最初にソースファイルの削除が失敗すると、削除ジョブは再試行されません。
  - [転送オプション] セクションで、次の操作を行います。
    - [All Formats] セクション:
      - [許可されているエラー数] には、BigQuery がジョブの実行中に無視できる不良レコードの最大数を入力します。不良レコード数がこの値を超えると、ジョブ結果で「invalid」が返されてジョブが失敗します。デフォルト値は 0 です。
    - [JSON, CSV] セクション:
      - 転送で宛先テーブルのスキーマに適合しないデータが削除されるようにする場合は、[不明な値を無視する] をオンにします。
    - [CSV] セクション:
      - [フィールド区切り文字] に、フィールドを区切る文字を入力します。デフォルト値はカンマ（,）です。
      - [スキップするヘッダー行] に、インポートの対象外にするソースファイル内のヘッダー行の数を入力します。デフォルト値は 0 です。
      - 引用符で囲まれたフィールド内で改行を許可するには、[引用符で囲まれた改行を許可する] をオンにします。
      - NULLABLE 列がない行を転送できるようにする場合は、[ジャグ行を許可する] をオンにします。
      注: 詳細については、CSV 専用オプションをご覧ください。
- （省略可）[通知オプション] セクションで、次の操作を行います。
  - 切り替えボタンをクリックしてメール通知を有効にします。このオプションを有効にすると、転送の実行が失敗した場合、転送管理者にメール通知が送信されます。
  - [Pub/Sub トピックを選択してください] で、トピック名を選択するか、[トピックを作成する] をクリックします。このオプションで、Pub/Sub の転送実行通知を構成します。
[保存] をクリックします。

bq

bq mk コマンドを入力して、転送作成フラグ --transfer_config を指定します。次のフラグも必要です。

--data_source
--display_name
--target_dataset
--params

bq mk \
--transfer_config \
--project_id=project_id \
--data_source=data_source \
--display_name=name \
--target_dataset=dataset \
--params='parameters'

ここで

project_id はプロジェクト ID です。--project_id で特定のプロジェクトを指定しない場合は、デフォルトのプロジェクトが使用されます。
data_source は、データソース（google_cloud_storage）です。
name は、転送構成の表示名です。転送名には、後で修正が必要になった場合に簡単に識別できる任意の名前を使用できます。
dataset は、転送構成の抽出先データセットです。
parameters には、作成される転送構成のパラメータを JSON 形式で指定します。例: --params='{"param":"param_value"}'。
- Cloud Storage では、data_path_template、destination_table_name_template、file_format パラメータを指定する必要があります。data_path_template は、転送するファイルが格納されている Cloud Storage URI です。ワイルドカードを 1 つ含めることができます。destination_table_name_template は、宛先テーブルの名前です。file_format には、転送するファイルのタイプ（CSV、JSON、AVRO、PARQUET、ORC）を指定します。デフォルト値は CSV です。
- すべての file_format 値について、オプションのパラメータとして max_bad_records を含めることができます。デフォルト値は 0 です。
- file_format の値として JSON または CSV を指定した場合は、オプションのパラメータとして ignore_unknown_values を含めることができます。file_format の値として CSV または JSON を指定していない場合、このパラメータは無視されます。
- file_format の値として CSV を指定した場合は、オプションのパラメータとして field_delimiter を含めて、フィールドを区切る文字を指定できます。デフォルト値はカンマ（,）です。file_format に CSV を選択していない場合、このパラメータは無視されます。
- file_format の値として CSV を指定した場合は、オプションのパラメータとして skip_leading_rows を含めて、インポートの対象外とするヘッダー行数を指定できます。デフォルト値は 0 です。file_format に CSV を選択していない場合、このパラメータは無視されます。
- file_format の値として CSV を指定した場合、引用符で囲まれたフィールド内で改行を許可するには、オプションのパラメータとして allow_quoted_newlines を含めます。file_format に CSV を選択していない場合、このパラメータは無視されます。
- file_format の値として CSV を指定した場合、末尾のオプションの列が欠落している行を許可するには、オプションのパラメータとして allow_jagged_rows を含めます。欠落した値には NULL が入力されます。file_format に CSV を選択していない場合、このパラメータは無視されます。
- オプションの delete_source_files パラメータを指定すると、転送が完了するたびにソースファイルが削除されます（最初にソースファイルの削除が失敗すると、削除ジョブは再試行されません）。delete_source_files のデフォルト値は false です。

たとえば、次のコマンドは、data_path_template の値に gs://mybucket/myfile/*.csv を使用し、ターゲットデータセットとして mydataset、file_format として CSV を指定した、My Transfer という名前の Cloud Storage 転送を作成します。この例では、file_format の CSV 値に関連するオプションのパラメータとして、デフォルト以外の値を使用しています。

この転送はデフォルトのプロジェクトで作成されます。

bq mk --transfer_config \
--target_dataset=mydataset \
--display_name='My Transfer' \
--params='{"data_path_template":"gs://mybucket/myfile/*.csv",
"destination_table_name_template":"MyTable",
"file_format":"CSV",
"max_bad_records":"1",
"ignore_unknown_values":"true",
"field_delimiter":"|",
"skip_leading_rows":"1",
"allow_quoted_newlines":"true",
"allow_jagged_rows":"false",
"delete_source_files":"true"}' \
--data_source=google_cloud_storage

コマンドを実行すると、次のようなメッセージが表示されます。

[URL omitted] Please copy and paste the above URL into your web browser and follow the instructions to retrieve an authentication code.

指示に従って、認証コードをコマンドラインに貼り付けます。

API

projects.locations.transferConfigs.create メソッドを使用して、TransferConfig リソースのインスタンスを指定します。

Java

GitHub で表示フィードバック

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.CreateTransferConfigRequest;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.ProjectName;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import com.google.protobuf.Struct;
import com.google.protobuf.Value;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

// Sample to create google cloud storage transfer config
public class CreateCloudStorageTransfer {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    final String projectId = "MY_PROJECT_ID";
    String datasetId = "MY_DATASET_ID";
    String tableId = "MY_TABLE_ID";
    // GCS Uri
    String sourceUri = "gs://cloud-samples-data/bigquery/us-states/us-states.csv";
    String fileFormat = "CSV";
    String fieldDelimiter = ",";
    String skipLeadingRows = "1";
    Map<String, Value> params = new HashMap<>();
    params.put(
        "destination_table_name_template", Value.newBuilder().setStringValue(tableId).build());
    params.put("data_path_template", Value.newBuilder().setStringValue(sourceUri).build());
    params.put("write_disposition", Value.newBuilder().setStringValue("APPEND").build());
    params.put("file_format", Value.newBuilder().setStringValue(fileFormat).build());
    params.put("field_delimiter", Value.newBuilder().setStringValue(fieldDelimiter).build());
    params.put("skip_leading_rows", Value.newBuilder().setStringValue(skipLeadingRows).build());
    TransferConfig transferConfig =
        TransferConfig.newBuilder()
            .setDestinationDatasetId(datasetId)
            .setDisplayName("Your Google Cloud Storage Config Name")
            .setDataSourceId("google_cloud_storage")
            .setParams(Struct.newBuilder().putAllFields(params).build())
            .setSchedule("every 24 hours")
            .build();
    createCloudStorageTransfer(projectId, transferConfig);
  }

  public static void createCloudStorageTransfer(String projectId, TransferConfig transferConfig)
      throws IOException {
    try (DataTransferServiceClient client = DataTransferServiceClient.create()) {
      ProjectName parent = ProjectName.of(projectId);
      CreateTransferConfigRequest request =
          CreateTransferConfigRequest.newBuilder()
              .setParent(parent.toString())
              .setTransferConfig(transferConfig)
              .build();
      TransferConfig config = client.createTransferConfig(request);
      System.out.println("Cloud storage transfer created successfully :" + config.getName());
    } catch (ApiException ex) {
      System.out.print("Cloud storage transfer was not created." + ex.toString());
    }
  }
}

転送のリフレッシュ実行の設定

Cloud Storage からの繰り返しの転送の設定に加えて、追加のデータファイルを転送の対象とするリフレッシュ実行を設定できます。

転送構成が日付に関連付けられている（パラメータ化されている）か Cloud Storage URI がパラメータ化されている場合、あるいはその両方が当てはまる場合、特定の日付について転送を実行できます。

転送のリフレッシュ実行を設定するには:

Console

Cloud Console の BigQuery ページに移動します。

[BigQuery] ページに移動
[転送] をクリックします。
目的の転送をクリックします。
[展開] メニューをクリックし、[更新転送] を選択します。
[バックフィル実行のスケジュール] ダイアログの [開始日] と [終了日] で日付を選択します。

注: Cloud Storage の転送構成がパラメータ化されていない場合、[更新転送] をクリックしても日付オプションは表示されません。この場合、[スケジュール] をクリックすると直ちに更新が行われます。

ファイルが更新転送の対象になるには、そのファイルが作成後 1 時間以上経過している必要があります。

次のステップ

Cloud Storage の転送でのランタイムパラメータの使用をご覧ください。
BigQuery Data Transfer Service の詳細をご覧ください。