Blob Storage の転送

Azure Blob Storage 用 BigQuery Data Transfer Service を使用すると、Blob Storage から BigQuery への定期的な読み込みジョブを自動的にスケジュールし、管理できます。

始める前に

Blob Storage の転送を作成する前に、次のことを行います。

BigQuery Data Transfer Service を有効にするために必要なすべての操作が完了していることを確認します。
既存の BigQuery データセットを選択するか、新しいデータセットを作成してデータを保存します。
データ転送用に既存の BigQuery テーブルを選択するか、新しい宛先テーブルを作成して、スキーマ定義を指定します。宛先テーブルは、テーブルの命名規則に従う必要があります。宛先テーブル名でもパラメータがサポートされています。
Blob Storage ストレージアカウント名、コンテナ名、データパス（省略可）、SAS トークンを取得します。Shared Access Signature（SAS）を使用して Blob Storage へのアクセス権を付与する方法については、Shared Access Signature（SAS）をご覧ください。
Azure Storage ファイアウォールを使用して Azure リソースへのアクセスを制限する場合は、許可された IP のリストに BigQuery Data Transfer Service ワーカーを追加します。
顧客管理の暗号鍵（CMEK）を指定する場合は、サービスアカウントに暗号化と復号の権限があることと、CMEK の使用に必要なCloud KMS 鍵のリソース ID があることを確認してください。CMEK が BigQuery Data Transfer Service とどのように連携するかについては、転送で暗号鍵を指定するをご覧ください。

必要な権限

Blob Storage のデータ転送を作成するには、bigquery.transfers.update Identity and Access Management（IAM）権限が必要です。また、ターゲットデータセットに対する bigquery.datasets.get 権限と bigquery.datasets.update 権限が必要です。

IAM 事前定義ロール bigquery.admin には、Blob Storage のデータ転送を作成するために必要な権限が含まれています。

BigQuery IAM の詳細については、IAM でのアクセス制御をご覧ください。

データ転送を有効にするために Blob Storage に適切な権限があることを確認するには、Shared Access Signature（SAS）をご覧ください。

Pub/Sub の転送実行通知を設定する場合は、pubsub.topics.setIamPolicy 権限が必要です。メール通知のみの場合は、Pub/Sub の権限は必要ありません。詳細については、BigQuery Data Transfer Service の実行通知をご覧ください。

制限事項

Blob Storage のデータ転送には、次の制限があります。

定期的なデータ転送の最小間隔は 1 時間です。デフォルトの間隔は 24 時間です。
Blob Storage のソースデータの形式によっては、追加の制限が適用される場合があります。
BigQuery Omni のロケーションへのデータ転送はサポートされていません。

Blob Storage データ転送を設定する

次のオプションのいずれかを選択します。

コンソール

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

[データ転送] に移動
[転送を作成] をクリックします。
[転送を作成] ページで、次の操作を行います。
- [ソースタイプ] セクションの [ソース] で、[Azure Blob Storage] を選択します。
- [転送構成名] セクションの [表示名] に、データ転送の名前を入力します。
- [スケジュールオプション] セクションで:
  - [繰り返しの頻度] を選択します。[時間]、[日]、[週]、[月] を選択する場合は、頻度も指定する必要があります。[カスタム] を選択して、カスタムの繰り返しの頻度を指定することもできます。[オンデマンド] を選択した場合、手動で転送をトリガーすると、このデータ転送が実行されます。
  - 必要に応じて、[すぐに開始可能] を選択するか、[設定した時刻に開始] を選択して開始日と実行時間を指定します。
- [転送先の設定] セクションの [データセット] で、データを保存するために作成したデータセットを選択します。
- [データソースの詳細] セクションで、次のようにします。
  - [宛先テーブル] に、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 をご覧ください。
- [転送オプション] セクションで、次の操作を行います。
  - [許容されるエラー数] に、無視できる不良レコードの最大数にあたる整数値を入力します。デフォルト値は 0 です。
  - （省略可）[Decimal target type] に、ソースデータの 10 進数値を変換できる SQL データ型のカンマ区切りのリストを入力します。変換にどの SQL データ型を選択するかは、次の条件によって決まります。
    - 型が指定されたリストに含まれていて、精度とスケールをサポートしている場合に、NUMERIC、BIGNUMERIC、STRING の順序で型が選択されます。
    - 表示されたデータ型のいずれも精度とスケールをサポートしていない場合は、指定したリストで最も広い範囲をサポートしているデータ型が選択されます。ソースデータの読み取り時に、値がサポートされている範囲を超えると、エラーがスローされます。
    - データ型 STRING では、すべての精度とスケールの値がサポートされています。
    - このフィールドを空のままにすると、ORC ではデータ型はデフォルトで NUMERIC,STRING になり、他のファイル形式では NUMERIC になります。
    - このフィールドに重複するデータ型を含めることはできません。
    - リストするデータ型の順序は無視されます。
- ファイル形式として CSV または JSON を選択した場合、スキーマに適合しない値を含んだ行を許容するには、[JSON, CSV] セクションで [不明な値を無視] をオンにします。
- ファイル形式として CSV を選択した場合は、[CSV] セクションでデータを読み込むための追加の CSV オプションを入力します。
- [通知オプション] セクションで、メール通知と Pub/Sub 通知を有効にするかどうか選択できます。
  - メール通知を有効にすると、転送の実行が失敗した場合、転送管理者にメール通知が送信されます。
  - Pub/Sub 通知を有効にするときに、公開するトピック名を選択するか、または [トピックを作成する] クリックします。
- CMEK を使用する場合は、[詳細オプション] セクションで [顧客管理の暗号鍵] を選択します。使用可能な CMEK のリストが表示され、ここから選択できます。CMEK が BigQuery Data Transfer Service と連携する仕組みについては、転送で暗号鍵を指定するをご覧ください。
[保存] をクリックします。

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

このサンプルを試す前に、クライアントライブラリを使用した BigQuery クイックスタートにある Java の設定手順を完了してください。詳細については、BigQuery Java API のリファレンスドキュメントをご覧ください。

BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアントライブラリの認証を設定するをご覧ください。


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

転送に使用する暗号鍵を指定する

転送実行のデータを暗号化する顧客管理の暗号鍵（CMEK）を指定できます。CMEK を使用して、Azure Blob Storage からの転送をサポートできます。

転送で CMEK を指定すると、BigQuery Data Transfer Service は取り込まれたデータの中間ディスク上キャッシュに CMEK を適用して、データ転送ワークフロー全体が CMEK 準拠になるようにします。

最初に CMEK で作成されていなかった既存の転送は、更新して CMEK を追加することはできません。たとえば、最初にデフォルトの方法で暗号化されていた宛先テーブルを CMEK で暗号化されるように変更することはできません。逆に、CMEK で暗号化された宛先テーブルを別のタイプの暗号化に変更することはできません。

転送構成が CMEK 暗号化を使用して最初に作成された場合は、転送の CMEK を更新できます。転送構成の CMEK を更新すると、BigQuery Data Transfer Service は転送の次回実行時に CMEK を宛先テーブルに伝播します。BigQuery Data Transfer Service は、古い CMEK を転送中に新しい CMEK に置き換えます。詳細については、転送の更新をご覧ください。

プロジェクトのデフォルト鍵を使用することもできます。転送でプロジェクトのデフォルト鍵を指定すると、BigQuery Data Transfer Service は、新しい転送構成のデフォルト鍵としてプロジェクトのデフォルト鍵を使用します。

転送の設定に関するトラブルシューティング

転送の設定に問題がある場合は、Blob Storage のデータ転送に関する問題をご覧ください。

次のステップ

転送のランタイムパラメータについて学習する。
BigQuery Data Transfer Service の詳細を確認する。
クロスクラウドの操作でデータを読み込む方法を学習する。