Blob Storage の転送

コンソール

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

   [データ転送] に移動

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

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

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

     

   • [転送構成名] セクションの [表示名] に、データ転送の名前を入力します。

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

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

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

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

   • [データソースの詳細] セクションで、次のようにします。

     • [Destination table] に、BigQuery でデータを保存するために作成したテーブルの名前を入力します。宛先テーブルの名前では、パラメータがサポートされています。
     • [Azure ストレージ アカウント名] に、blob ストレージ アカウント名を入力します。
     • [コンテナ名] に、Blob Storage のコンテナ名を入力します。
     • [データパス] に、転送するファイルをフィルタするためのパスを入力します。例をご覧ください。
     • [SAS トークン] に、Azure SAS トークンを入力します。
     • [ファイル形式] でソースデータ形式を選択します。
     • [書き込み処理] で、[WRITE_APPEND] を選択して宛先テーブルに新しいデータを段階的に追加するか、[WRITE_TRUNCATE] を選択して各転送の実行中に宛先テーブル内のデータを上書きします。WRITE_APPEND は、[書き込み処理] のデフォルト値です。

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

     

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

     • [Number of errors allowed] に、無視できる不良レコードの最大数にあたる整数値を入力します。 デフォルト値は 0 です。
     • （省略可）[Decimal target type] に、ソースデータの 10 進数値を変換できる SQL データ型のカンマ区切りのリストを入力します。変換にどの SQL データ型を選択するかは、次の条件によって決まります。
       • 型が指定されたリストに含まれていて、精度とスケールをサポートしている場合に、NUMERIC、BIGNUMERIC、STRING の順序で型が選択されます。
       • 表示されたデータ型のいずれも精度とスケールをサポートしていない場合は、指定したリストで最も広い範囲をサポートしているデータ型が選択されます。ソースデータの読み取り時に、値がサポートされている範囲を超えると、エラーがスローされます。
       • データ型 STRING では、すべての精度とスケールの値がサポートされています。
       • このフィールドを空のままにすると、ORC ではデータ型はデフォルトで NUMERIC,STRING になり、他のファイル形式では NUMERIC になります。
       • このフィールドに重複するデータ型を含めることはできません。
       • リストするデータ型の順序は無視されます。

   • ファイル形式として CSV または JSON を選択した場合、スキーマに適合しない値を含んだ行を許容するには、[JSON, CSV] セクションで [Ignore unknown values] をオンにします。

   • ファイル形式として CSV を選択した場合は、[CSV] セクションでデータを読み込むための追加の CSV オプションを入力します。

     

   • [通知オプション] セクションで、メール通知と Pub/Sub 通知を有効にするかどうか選択できます。

     • メール通知を有効にすると、転送の実行が失敗した場合、転送管理者にメール通知が送信されます。
     • Pub/Sub 通知を有効にするときに、公開するトピック名を選択するか、または [トピックを作成する] クリックします。

   • CMEK を使用する場合は、[詳細オプション] セクションで [顧客管理の暗号鍵] を選択します。使用可能な CMEK のリストが表示され、ここから選択できます。CMEK が BigQuery Data Transfer Service と連携する仕組みについては、転送で暗号鍵を指定するをご覧ください。

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

bq

bq mk --transfer_config コマンドを使用して、Blob Storage の転送を作成します。

bq mk \
  --transfer_config \
  --project_id=PROJECT_ID \
  --data_source=DATA_SOURCE \
  --display_name=DISPLAY_NAME \
  --target_dataset=DATASET \
  --destination_kms_key=DESTINATION_KEY \
  --params=PARAMETERS

以下を置き換えます。

• PROJECT_ID: 省略可。ターゲット データセットを含むプロジェクト ID。指定しない場合は、デフォルトのプロジェクトが使用されます。
• DATA_SOURCE: azure_blob_storage
• DISPLAY_NAME: データ転送構成の表示名。転送名には、後で修正が必要になった場合に識別できる任意の名前を使用できます。
• DATASET: データ転送構成のターゲット データセット。
• DESTINATION_KEY:（省略可）Cloud KMS 鍵のリソース ID（例: projects/project_name/locations/us/keyRings/key_ring_name/cryptoKeys/key_name）。
• PARAMETERS: JSON 形式で一覧表示されるデータ転送構成のパラメータ。例: --params={"param1":"value1", "param2":"value2"}。Blob Storage データ転送のパラメータは次のとおりです。
  • destination_table_name_template: 必須。宛先テーブルの名前を指定します。
  • storage_account: 必須。Blob Storage アカウント名。
  • container: 必須。Blob Storage のコンテナ名。
  • data_path: 省略可。転送するファイルをフィルタするパス。例をご覧ください。
  • sas_token: 必須。Azure SAS トークン。
  • file_format: 省略可。転送するファイルの種類（CSV、JSON、AVRO、PARQUET、ORC）。デフォルト値は CSV です。
  • write_disposition: 省略可。宛先テーブルにデータを追加する場合、WRITE_APPEND を選択します。宛先テーブルのデータを上書きする場合は、WRITE_TRUNCATE を選択します。デフォルト値は WRITE_APPEND です。
  • max_bad_records: 省略可。許可する不良レコードの数を指定します。デフォルト値は 0 です。
  • decimal_target_types: 省略可。ソースデータの 10 進数値を変換できる、有効な SQL データ型のカンマ区切りのリスト。このフィールドを指定しない場合は、ORC ではデータ型はデフォルトで NUMERIC,STRING になり、他のファイル形式では NUMERIC になります。
  • ignore_unknown_values: 省略可。file_format が JSON または CSV でない場合は無視されます。スキーマに適合しない値を含む行を許可するには、true に設定します。
  • field_delimiter: 省略可。file_format が CSV の場合にのみ適用されます。フィールドを区切る文字を指定します。デフォルト値は , です。
  • skip_leading_rows: 省略可。file_format が CSV の場合にのみ適用されます。インポートの対象外にするヘッダー行の数を指定します。デフォルト値は 0 です。
  • allow_quoted_newlines: 省略可。file_format が CSV の場合にのみ適用されます。引用符で囲まれたフィールド内で改行を許可するかどうかを指定します。
  • allow_jagged_rows: 省略可。file_format が CSV の場合にのみ適用されます。末尾のオプションの列が欠落している行を許可するかどうかを指定します。欠損値には NULL が入力されます。

たとえば、以下では mytransfer という Blob Storage のデータ転送を作成します。

bq mk \
  --transfer_config \
  --data_source=azure_blob_storage \
  --display_name=mytransfer \
  --target_dataset=mydataset \
  --destination_kms_key=projects/myproject/locations/us/keyRings/mykeyring/cryptoKeys/key1
  --params={"destination_table_name_template":"mytable",
      "storage_account":"myaccount",
      "container":"mycontainer",
      "data_path":"myfolder/*.csv",
      "sas_token":"my_sas_token_value",
      "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"}

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 azure blob storage transfer config.
public class CreateAzureBlobStorageTransfer {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    final String projectId = "MY_PROJECT_ID";
    final String displayName = "MY_TRANSFER_DISPLAY_NAME";
    final String datasetId = "MY_DATASET_ID";
    String tableId = "MY_TABLE_ID";
    String storageAccount = "MY_AZURE_STORAGE_ACCOUNT_NAME";
    String containerName = "MY_AZURE_CONTAINER_NAME";
    String dataPath = "MY_AZURE_FILE_NAME_OR_PREFIX";
    String sasToken = "MY_AZURE_SAS_TOKEN";
    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("storage_account", Value.newBuilder().setStringValue(storageAccount).build());
    params.put("container", Value.newBuilder().setStringValue(containerName).build());
    params.put("data_path", Value.newBuilder().setStringValue(dataPath).build());
    params.put("sas_token", Value.newBuilder().setStringValue(sasToken).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());
    createAzureBlobStorageTransfer(projectId, displayName, datasetId, params);
  }

  public static void createAzureBlobStorageTransfer(
      String projectId, String displayName, String datasetId, Map<String, Value> params)
      throws IOException {
    TransferConfig transferConfig =
        TransferConfig.newBuilder()
            .setDestinationDatasetId(datasetId)
            .setDisplayName(displayName)
            .setDataSourceId("azure_blob_storage")
            .setParams(Struct.newBuilder().putAllFields(params).build())
            .setSchedule("every 24 hours")
            .build();
    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests.
    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("Azure Blob Storage transfer created successfully: " + config.getName());
    } catch (ApiException ex) {
      System.out.print("Azure Blob Storage transfer was not created." + ex.toString());
    }
  }
}