Cloud Storage の転送

コンソール

1. Google Cloud コンソールの [データ転送] ページに移動します。

   [データ転送] に移動

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

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

   

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

   

5. [スケジュール オプション] セクションで:

   • [繰り返しの頻度] を選択します。[時間]、[日]、[週]、[月] を選択する場合は、頻度も指定する必要があります。[カスタム] を選択して、カスタムの繰り返しの頻度を指定することもできます。[オンデマンド] を選択した場合、手動で転送をトリガーすると、このデータ転送が実行されます。

   • 必要に応じて、[すぐに開始可能] を選択するか、[設定した時刻に開始] を選択して開始日と実行時間を指定します。

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

   

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

   1. [宛先テーブル] に宛先テーブルの名前を入力します。宛先テーブルは、テーブルの命名規則に従う必要があります。宛先テーブル名でもパラメータがサポートされています。
   2. [Cloud Storage URI] に Cloud Storage URI を入力します。ワイルドカードとパラメータがサポートされています。
   3. [書き込み設定] の場合、次から選択します。
      • APPEND で、既存の宛先テーブルに新しいデータを追加します。APPEND は、[書き込み設定] のデフォルト値です。
      • MIRROR で、各データ転送の実行中に宛先テーブル内のデータを上書きします。

   BigQuery Data Transfer Service が APPEND または MIRROR を使用してデータを取り込む方法の詳細については、Cloud Storage の転送データの取り込みをご覧ください。writeDisposition フィールドの詳細については、JobConfigurationLoad をご覧ください。

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

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

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

      4. [CSV] セクション:

         1. [フィールド区切り文字] に、フィールドを区切る文字を入力します。デフォルト値はカンマ（,）です。
         2. [引用符文字] に、CSV ファイル内のデータ セクションを囲む引用符として使用する文字を入力します。デフォルト値は二重引用符（"）です。
         3. [スキップするヘッダー行] に、インポートの対象外にするソースファイル内のヘッダー行の数を入力します。デフォルト値は 0 です。
         4. 引用符で囲まれたフィールド内で改行を許可するには、[引用符で囲まれた改行を許可する] をオンにします。
         5. NULLABLE 列がない行をデータ転送できるようにする場合は、[ジャグ行を許可する] をオンにします。

8. [サービス アカウント] メニューで、Google Cloud プロジェクトに関連付けられたサービス アカウントからサービス アカウントを選択します。ユーザー認証情報を使用する代わりに、サービス アカウントをデータ転送に関連付けることができます。データ転送でサービス アカウントを使用する方法の詳細については、サービス アカウントの使用をご覧ください。

   • フェデレーション ID でログインした場合、データ転送を作成するにはサービス アカウントが必要です。Google アカウントでログインした場合、データ転送用のサービス アカウントは省略可能です。
   • サービス アカウントには、BigQuery と Cloud Storage の両方に対して必要な権限が付与されている必要があります。

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

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

10. 省略可: [詳細オプション] セクションで、次の操作を行います。

    • CMEK を使用する場合は、[顧客管理の暗号鍵] を選択します。使用可能な CMEK のリストが表示され、ここから選択できます。

    CMEK が BigQuery Data Transfer Service と連携する仕組みについては、転送で暗号鍵を指定するをご覧ください。

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

bq

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

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

オプションのフラグ:

• --destination_kms_key: このデータ転送に顧客管理の暗号鍵（CMEK）を使用する場合、Cloud KMS 鍵の鍵のリソース ID を指定します。CMEK が BigQuery Data Transfer Service と連携する仕組みについては、転送で暗号鍵を指定するをご覧ください。
• --service_account_name: ユーザー アカウントの代わりに、Cloud Storage の転送認証に使用するサービス アカウントを指定します。

bq mk \
--transfer_config \
--project_id=PROJECT_ID \
--data_source=DATA_SOURCE \
--display_name=NAME \
--target_dataset=DATASET \
--destination_kms_key="DESTINATION_KEY" \
--params='PARAMETERS' \
--service_account_name=SERVICE_ACCOUNT_NAME

ここで

• PROJECT_ID はプロジェクト ID です。--project_id で特定のプロジェクトを指定しない場合は、デフォルトのプロジェクトが使用されます。
• DATA_SOURCE はデータソースです（例: google_cloud_storage）。
• NAME は、データ転送構成の表示名です。転送名には、後で修正が必要になった場合に識別できる任意の名前を使用できます。
• DATASET は、転送構成の抽出先データセットです。
• DESTINATION_KEY: Cloud KMS 鍵のリソース ID（例: projects/project_name/locations/us/keyRings/key_ring_name/cryptoKeys/key_name）。
• PARAMETERS には、作成される転送構成のパラメータを JSON 形式で指定します。（例: --params='{"param":"param_value"}'）。
  • destination_table_name_template: 宛先 BigQuery テーブルの名前。
  • data_path_template: 転送するファイルが格納されている Cloud Storage URI です。ワイルドカードを 1 つ含めることができます。
  • write_disposition: 一致するファイルを宛先テーブルに追加するか、完全にミラーリングするかを決定します。サポートされている値は、APPEND と MIRROR です。BigQuery Data Transfer Service が Cloud Storage の転送でデータを追加またはミラーリングする方法については、Cloud Storage 転送のデータの取り込みをご覧ください。
  • file_format: 転送するファイルの形式。形式は、CSV、JSON、AVRO、PARQUET、ORC のいずれかです。デフォルト値は CSV です。
  • max_bad_records: 任意の file_format 値について、無視できる不良レコードの最大数。デフォルト値は 0 です。
  • decimal_target_types: 任意の file_format 値に対する、ソースの 10 進数値を変換できる、有効な SQL データ型のカンマ区切りのリスト。このフィールドを指定しない場合は、ORC ではデータ型はデフォルトで "NUMERIC,STRING" になり、他のファイル形式では "NUMERIC" になります。
  • ignore_unknown_values: 任意の file_format の値に対して TRUE に設定すると、スキーマに一致しない値を含む行が許容されます。詳細については、JobConfigurationLoad 参照テーブルの ignoreUnknownvalues フィールドの詳細をご覧ください。
  • use_avro_logical_types: AVRO file_format の値に対して TRUE に設定すると、元の型（例: INTEGER）のみを使用するのではなく、論理型を対応する型に変換します（例: TIMESTAMP）。
  • parquet_enum_as_string: PARQUET file_format の値に対して TRUE に設定すると、PARQUET ENUM 論理型を、デフォルトの BYTES ではなく STRING と推測します。
  • parquet_enable_list_inference: PARQUET file_format の値に対して TRUE に設定すると、PARQUET LIST 論理型専用のスキーマ推定を使用します。
  • reference_file_schema_uri: リーダー スキーマを含む参照ファイルへの URI パス。
  • field_delimiter: CSV file_format の値に対する、フィールドを区切る文字。デフォルト値はカンマ（,）です。
  • quote: CSV file_format の値に対して、CSV ファイル内のデータ セクションを引用符で囲むために使用される文字。デフォルト値は二重引用符（"）です。
  • skip_leading_rows: CSV file_format の値に対して、インポート対象外にする先頭ヘッダー行の数を指定します。デフォルト値は 0 です。
  • allow_quoted_newlines: CSV file_format の値に対して TRUE に設定すると、引用符で囲まれたフィールド内での改行が許可されます。
  • allow_jagged_rows: CSV file_format の値に対して TRUE に設定すると、末尾のオプションの列が欠落している行を受け入れます。欠損値には NULL が入力されます。
  • preserve_ascii_control_characters: CSV file_format の値に対して TRUE に設定すると、埋め込みの ASCII 制御文字を保持します。
  • encoding: CSV のエンコード タイプを指定します。サポートされる値は UTF8、ISO_8859_1、UTF16BE、UTF16LE、UTF32BE、UTF32LE です。
  • delete_source_files: TRUE に設定すると、転送が完了するたびにソースファイルを削除します。ソースファイルの削除の初めての試行が失敗した場合、削除ジョブは再実行されません。デフォルト値は FALSE です。
• SERVICE_ACCOUNT_NAME は、転送の認証に使用されるサービス アカウント名です。サービス アカウントは、転送の作成に使用した project_id が所有している必要があります。また、必要な権限がすべて付与されている必要があります。

たとえば、次のコマンドは、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 \
--project_id=myProject \
--display_name='My Transfer' \
--destination_kms_key=projects/myproject/locations/mylocation/keyRings/myRing/cryptoKeys/myKey \
--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":"|",
"quote":";",
"skip_leading_rows":"1",
"allow_quoted_newlines":"true",
"allow_jagged_rows":"false",
"delete_source_files":"true"}' \
--data_source=google_cloud_storage \
--service_account_name=abcdef-test-sa@abcdef-test.iam.gserviceaccount.com projects/862514376110/locations/us/transferConfigs/ 5dd12f26-0000-262f-bc38-089e0820fe38

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

[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());
    }
  }
}

コンソール

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

   [BigQuery] ページに移動

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

3. リストからデータ転送を選択します。

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

   • [今すぐ転送を実行] をクリックした場合は、[1 回限りの転送を実行] または [特定の日付で実行] を選択します（該当する場合）。[特定の日付で実行] を選択した場合は、具体的な日時を選択します。

     

   • [バックフィルをスケジュール] をクリックした場合は、[1 回限りの転送を実行] または [期間を指定して実行] を適宜選択します。[期間を指定して実行] を選択した場合は、開始日時と終了日時を選択します。

     

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

bq

bq mk コマンドを入力して、--transfer_run フラグを指定します。--run_time フラグ、または --start_time フラグと --end_time フラグを使用できます。

bq mk \
--transfer_run \
--start_time='START_TIME' \
--end_time='END_TIME' \
RESOURCE_NAME

bq mk \
--transfer_run \
--run_time='RUN_TIME' \
RESOURCE_NAME

ここで

• START_TIME と END_TIME は、Z で終わるタイムスタンプか、有効なタイムゾーンのオフセットを含むタイムスタンプです。次に例を示します。

  • 2017-08-19T12:11:35.00Z
  • 2017-05-25T00:00:00+00:00

• RUN_TIME は、データ転送実行をスケジュールする時間を指定するタイムスタンプです。現在の時刻に対して 1 回限りの転送を実行する場合は、--run_time フラグを使用できます。

• RESOURCE_NAME は、転送のリソース名（転送構成）です（例: projects/myproject/locations/us/transferConfigs/1234a123-1234-1a23-1be9-12ab3c456de7）。転送のリソース名がわからない場合は、bq ls --transfer_config --transfer_location=LOCATION コマンドを実行してリソース名を確認します。

API

projects.locations.transferConfigs.startManualRuns メソッドを使用し、parent パラメータで転送構成のリソース名を指定します。