Cloud Storage の転送

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

始める前に

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

制限事項

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

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

  • BigQuery Data Transfer Service では外部データソースのデータ整合性は保証されません。クエリの実行中に基になるデータを変更すると、予期しない動作が発生する可能性があります。

  • Cloud Storage のソースデータの形式によっては、追加の制限が適用される場合があります。詳細については、次をご覧ください。

  • Cloud Storage バケットは、BigQuery の宛先データセットのリージョンまたはマルチリージョンと互換性のあるリージョンまたはマルチリージョンに存在する必要があります。これはコロケーションと呼ばれます。詳細については、Cloud Storage 転送のデータのロケーションをご覧ください。

最小間隔

  • ソースファイルは、ファイル作成後の最小間隔なしで、すぐに取得され転送されます。
  • 定期的な転送の最小間隔は 15 分です。デフォルトの定期的な転送間隔は 24 時間です。

必要な権限

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

  • BigQuery: 転送を作成するユーザーに、BigQuery で次の権限が付与されていることを確認します。

    • bigquery.transfers.update（転送を作成する権限）
    • bigquery.datasets.get と bigquery.datasets.updateの両方（抽出先データセットに対する権限）

    bigquery.transfers.update 権限、bigquery.datasets.update 権限、bigquery.datasets.get 権限は 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

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

    [BigQuery] ページに移動

  2. [転送] をクリックします。

  3. [作成] をクリックします。

  4. [転送の作成] ページで、次の操作を行います。

    • [ソースタイプ] セクションで、[ソース] として [Cloud Storage] を選択します。

      転送のソース

    • [転送構成名] セクションの [表示名] に、転送名（例: My Transfer）を入力します。転送名には、後で修正が必要になった場合に簡単に識別できる任意の名前を使用できます。

      転送名

    • [スケジュール オプション] セクションで、[スケジュール] をデフォルト値（[すぐに開始可能]）のままにするか、[設定した時間に開始] をクリックします。

      • [繰り返しの頻度] で、転送を実行する頻度のオプションを選択します。最小間隔は 15 分です。
        • 毎日（デフォルト）
        • 毎週
        • 毎月
        • カスタム[カスタム スケジュール] にカスタム頻度を入力します。例: every day 00:00。スケジュールのフォーマットをご覧ください。
        • オンデマンド

      • [開始日と実行時間] に、転送を開始する日付と時刻を入力します。[すぐに開始可能] を選択した場合、このオプションは無効になります。

        転送スケジュール

    • [転送先の設定] セクションの [宛先データセット] で、データを保存するために作成したデータセットを選択します。

      転送データセット

    • [データソースの詳細] セクションで、次の操作を行います。

      • [宛先テーブル] に宛先テーブルの名前を入力します。宛先テーブルは、テーブルの命名規則に従う必要があります。宛先テーブル名でもパラメータがサポートされています。
      • [Cloud Storage URI] に Cloud Storage URI を入力します。ワイルドカードとパラメータがサポートされています。

      • [書き込み設定] の場合、次から選択します。

        • APPEND で、既存の宛先テーブルに新しいデータを追加する。
        • MIRROR で、宛先テーブル内のデータを更新し、変更されたデータをソースに反映する。MIRROR は、宛先テーブル内のデータの新しいコピーを上書きします。

      • 転送が完了するたびにソースファイルを削除する場合は、[Delete source files after transfer] をオンにします。削除ジョブはベスト エフォート ベースです。最初にソースファイルの削除が失敗すると、削除ジョブは再試行されません。

      • [転送オプション] セクションで、次の操作を行います。

        • [All Formats] セクション:
          • [許可されているエラー数] には、BigQuery がジョブの実行中に無視できる不良レコードの最大数を入力します。不良レコード数がこの値を超えると、ジョブ結果で「invalid」が返されてジョブが失敗します。デフォルト値は 0 です。
          • （省略可）[Decimal target type] に、ソースの 10 進数値を変換できる、有効な SQL データ型のカンマ区切りのリストを入力します。変換にどの SQL データ型を選択するかは、次の条件によって決まります。
            • 変換に選択したデータ型は、ソースデータの精度とスケールをサポートする次のリスト内の最初のデータ型になります。順序は NUMERIC、BIGNUMERIC、STRING の順になります。
            • リストされたデータ型が精度とスケールをサポートしていない場合は、指定されたリストの最も広い範囲をサポートするデータ型が選択されます。ソースデータの読み取り時に、値がサポートされている範囲を超えると、エラーがスローされます。
            • データ型 STRING では、すべての精度とスケールの値がサポートされています。
            • このフィールドが空の場合、ORC ではデータ型はデフォルトで「NUMERIC、STRING」、その他のファイル形式では「NUMERIC」になります。
            • このフィールドには、重複するデータタイプを含めることはできません。
            • このフィールドでリストするデータ型の順序は無視されます。
        • [JSON, CSV] セクション:
          • 転送で宛先テーブルのスキーマに適合しないデータが削除されるようにする場合は、[不明な値を無視する] をオンにします。

        • [CSV] セクション:

          • [フィールド区切り文字] に、フィールドを区切る文字を入力します。デフォルト値はカンマ（,）です。
          • [スキップするヘッダー行] に、インポートの対象外にするソースファイル内のヘッダー行の数を入力します。デフォルト値は 0 です。
          • 引用符で囲まれたフィールド内で改行を許可するには、[引用符で囲まれた改行を許可する] をオンにします。
          • NULLABLE 列がない行を転送できるようにする場合は、[ジャグ行を許可する] をオンにします。

      Cloud Storage ソースの詳細

    • （省略可）[通知オプション] セクションで、次の操作を行います。

      • 切り替えボタンをクリックしてメール通知を有効にします。このオプションを有効にすると、転送の実行が失敗した場合、転送管理者にメール通知が送信されます。
      • [Pub/Sub トピックを選択してください] で、トピック名を選択するか、[トピックを作成する] をクリックします。このオプションで、Pub/Sub の転送実行通知を構成します。

  5. [保存] をクリックします。

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 値について、オプションのパラメータとして decimal_target_types を含めることができます。decimal_target_types は、ソースの 10 進数値を変換できる、有効な SQL データ型のカンマ区切りのリストです。このフィールドが指定されていない場合、ORC ではデータ型はデフォルトで「NUMERIC、STRING」、その他のファイル形式では「NUMERIC」になります。
    • 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

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 から自動で転送がスケジュールされるだけでなく、転送を手動でトリガーして追加のデータファイルを読み込むこともできます。

転送の構成がパラメータ化されたランタイムである場合、追加の転送を開始する期間を指定する必要があります。

手動で転送をトリガーするには:

コンソール

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

    BigQuery ページに移動

  2. [データ転送] をクリックします。

  3. 目的の転送をクリックします。

  4. [今すぐ転送を実行] または [バックフィルのスケジュール構成] をクリックします（パラメータ化されたランタイムの転送構成の場合）。

  5. 必要に応じて、[開始日] と [終了日] を選択し、[OK] をクリックして確定します。

    今すぐ転送を実行

    パラメータ化されたランタイムの転送構成の場合、[バックフィルのスケジュール構成] をクリックすると日付オプションが表示されます。

    バックフィルのスケジュール構成

次のステップ