データのバッチ読み込み

BigQuery には、Cloud Storage またはローカル ファイルからバッチ オペレーションとしてデータを読み込むことができます。ソースデータは次のいずれかの形式になります。

• Avro
• カンマ区切り値（CSV）
• JSON（改行区切り）
• ORC
• Parquet
• Cloud Storage に保存されている Datastore のエクスポート
• Cloud Storage に保存されている Firestore のエクスポート

BigQuery Data Transfer Service を使用して、Cloud Storage から BigQuery への定期的な読み込みを設定することもできます。

始める前に

このドキュメントの各タスクを行うのに必要な権限をユーザーに与える Identity and Access Management（IAM）ロールを付与し、データを保存するためのデータセットを作成します。

必要な権限

BigQuery にデータを読み込むには、読み込みジョブを実行してデータを BigQuery のテーブルとパーティションに読み込む IAM 権限が必要です。Cloud Storage からデータを読み込む場合は、データを含むバケットに対する IAM アクセス権限も必要です。

BigQuery にデータを読み込む権限

新しい BigQuery テーブルやパーティションにデータを読み込む場合、または既存のテーブルやパーティションにデータの追加や上書きを行う場合は、次の IAM 権限が必要です。

• bigquery.tables.create
• bigquery.tables.updateData
• bigquery.tables.update
• bigquery.jobs.create

以下の各事前定義 IAM ロールには、BigQuery テーブルやパーティションにデータを読み込むために必要な権限が含まれています。

• roles/bigquery.dataEditor
• roles/bigquery.dataOwner
• roles/bigquery.admin（bigquery.jobs.create 権限を含む）
• bigquery.user（bigquery.jobs.create 権限を含む）
• bigquery.jobUser（bigquery.jobs.create 権限を含む）

また、bigquery.datasets.create 権限がある場合は、作成するデータセットで読み込みジョブを使用してテーブルの作成と更新を行えます。

BigQuery での IAM のロールと権限については、事前定義ロールと権限をご覧ください。

Cloud Storage からデータを読み込む権限

Cloud Storage バケットからデータを読み込むために必要な権限を取得するには、バケットに対するストレージ管理者（roles/storage.admin）IAM ロールを付与するよう管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

この事前定義ロールには、Cloud Storage バケットからデータを読み込むために必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

データセットを作成する

データを保存する BigQuery データセットを作成します。

Cloud Storage からのデータの読み込み

BigQuery では以下の Cloud Storage ストレージ クラスからデータを読み込むことができます。

• Standard
• Nearline
• Coldline
• アーカイブ

BigQuery にデータを読み込む方法については、以下のデータ形式のページをご覧ください。

• CSV
• JSON
• Avro
• Parquet
• ORC
• Datastore エクスポート
• Firestore エクスポート

Cloud Storage から BigQuery への定期的な読み込みを構成する方法については、Cloud Storage の転送をご覧ください。

ロケーションに関する留意事項

Cloud Storage からデータを読み込む場合、読み込まれるデータは BigQuery データセットと同じロケーションにある必要があります。

• BigQuery データセットが US マルチリージョンにある場合、任意のロケーションにある Cloud Storage バケットからデータを読み込むことができます。

• マルチリージョン バケット: 読み込み元の Cloud Storage バケットがマルチリージョン バケットにある場合、BigQuery データセットは同じマルチリージョン バケット、または同じマルチリージョン バケットに含まれる任意の単一リージョンに配置できます。たとえば、Cloud Storage バケットが EU リージョンにある場合、BigQuery データセットは EU マルチリージョンまたは EU の任意の単一リージョンに配置できます。

• デュアルリージョン バケット: 読み込み元の Cloud Storage バケットがデュアルリージョン バケットにある場合、BigQuery データセットは、デュアルリージョン バケットに含まれるリージョン、またはデュアルリージョンを含むマルチリージョンに配置できます。たとえば、Cloud Storage バケットが EUR4 リージョンにある場合、BigQuery データセットはフィンランド（europe-north1）の単一リージョン、オランダ（europe-west4）の単一リージョンまたは EU マルチリージョンに配置できます。

• 単一リージョン バケット: 読み込み元の Cloud Storage バケットが単一リージョンにある場合、BigQuery データセットは同じ単一リージョン、またはその単一リージョンを含むマルチリージョンに配置できます。たとえば、Cloud Storage バケットがフィンランド（europe-north1）リージョンにある場合、BigQuery データセットはフィンランドまたは EU マルチリージョンに配置できます。

• 1 つ例外があります。BigQuery データセットが asia-northeast1 リージョンにある場合、Cloud Storage バケットは EU マルチリージョンに配置できます。

Cloud Storage のロケーションについて詳しくは、Cloud Storage ドキュメントのバケットのロケーションをご覧ください。

データセットの作成後にそのロケーションを変更することはできませんが、データセットのコピーは作成できます。また、手動で移動することもできます。詳細については、次をご覧ください。

• データセットのコピー
• データセットの移動

Cloud Storage URI の取得

Cloud Storage データソースからデータを読み込むには、Cloud Storage URI を指定する必要があります。

Cloud Storage のリソースパスには、バケット名とオブジェクト（ファイル名）が含まれます。たとえば、Cloud Storage バケットの名前が mybucket でデータファイルの名前が myfile.csv の場合、リソースパスは gs://mybucket/myfile.csv になります。

BigQuery では、最初のダブル スラッシュの後に複数の連続スラッシュが含まれる Cloud Storage リソースパスはサポートされていません。Cloud Storage オブジェクトの名前には複数の連続スラッシュ（"/"）文字を含めることができます。一方、BigQuery では、複数の連続スラッシュは単一のスラッシュに変換されます。たとえば、リソースパス gs://bucket/my//object//name は Cloud Storage では有効ですが、BigQuery では機能しません。

Cloud Storage のリソースパスを取得するには:

1. Cloud Storage コンソールを開きます。

   Cloud Storage コンソール

2. ソースデータを含むオブジェクト（ファイル）の場所に移動します。

3. オブジェクトの名前をクリックします。

   [オブジェクトの詳細] ページが開きます。

4. [gsutil URI] フィールドに表示されている値（gs:// で始まる）をコピーします。

Google Datastore のエクスポートで指定できる URI は 1 つのみです。また、URI の末尾は .backup_info または .export_metadata である必要があります。

Cloud Storage の URI でのワイルドカードのサポート

データが複数のファイルに分かれている場合は、ワイルドカードとしてアスタリスク（*）を使用して複数のファイルを選択できます。ワイルドカードとしてアスタリスクを使用する場合は、次のルールに従う必要があります。

• アスタリスクは、オブジェクト名の中または末尾に使用できます。
• 複数のアスタリスクは使用できません。たとえば、パス gs://mybucket/fed-*/temp/*.csv は無効です。
• バケット名にはアスタリスクを使用できません。

例:

• 次の例では、gs://mybucket/fed-samples/fed-sample で始まるすべてのフォルダ内のすべてのファイルを選択する方法を示します。

  gs://mybucket/fed-samples/fed-sample*
  

• 次の例では、fed-samples というフォルダと fed-samples のサブフォルダにある .csv という拡張子のファイルのみを選択する方法を示します。

  gs://mybucket/fed-samples/*.csv
  

• 次の例では、fed-samples という名前のフォルダで fed-sample*.csv という命名パターンのファイルを選択する方法を示します。この例では、fed-samples のサブフォルダ内のファイルは選択されません。

  gs://mybucket/fed-samples/fed-sample*.csv
  

一部のプラットフォームでは、bp コマンドライン ツールの使用時に、アスタリスクをエスケープしなければならない場合があります。

Cloud Storage から Datastore または Firestore のエクスポート データを読み込むときには、アスタリスク ワイルドカードは使用できません。

制限事項

Cloud Storage バケットから BigQuery にデータを読み込む際には、次の制限があります。

• データセットのロケーションが US マルチリージョン以外の値に設定されている場合、Cloud Storage バケットはデータセットと同じリージョンに存在するか、同じマルチリージョンに含まれている必要があります。
• BigQuery では外部データソースに対して整合性が保証されません。クエリの実行中に基になるデータを変更すると、予期しない動作が発生する可能性があります。
• BigQuery では、Cloud Storage オブジェクトのバージョニングはサポートされていません。Cloud Storage URI に世代番号を含めると、読み込みジョブは失敗します。

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

• CSV に関する制限
• JSON に関する制限
• Datastore エクスポートに関する制限
• Firestore エクスポートに関する制限
• ネストされた繰り返しデータに関する制限

ローカル ファイルからのデータの読み込み

次のいずれかを使用して、読み取り可能なデータソース（ローカルマシンなど）からデータを読み込むことができます。

• Google Cloud コンソール
• bq コマンドライン ツールの bq load コマンド
• API
• クライアント ライブラリ

Google Cloud コンソールまたは bq コマンドライン ツールを使用してデータを読み込むと、読み込みジョブが自動的に作成されます。

データをローカル データソースから読み込むには:

bq --location=LOCATION load \
--source_format=FORMAT \
PROJECT_ID:DATASET.TABLE \
PATH_TO_SOURCE \
SCHEMA

    bq load \
    --source_format=NEWLINE_DELIMITED_JSON \
    mydataset.mytable \
    ./mydata.json \
    ./myschema.json

    bq load \
    --source_format=CSV \
    myotherproject:mydataset.mytable \
    ./mydata.csv \
    qtr:STRING,sales:FLOAT,year:STRING

    bq load \
    --autodetect \
    --source_format=CSV \
    mydataset.mytable \
    ./mydata.csv

using Google.Cloud.BigQuery.V2;
using System;
using System.IO;

public class BigQueryLoadFromFile
{
    public void LoadFromFile(
        string projectId = "your-project-id",
        string datasetId = "your_dataset_id",
        string tableId = "your_table_id",
        string filePath = "path/to/file.csv"
    )
    {
        BigQueryClient client = BigQueryClient.Create(projectId);
        // Create job configuration
        var uploadCsvOptions = new UploadCsvOptions()
        {
            SkipLeadingRows = 1,  // Skips the file headers
            Autodetect = true
        };
        using (FileStream stream = File.Open(filePath, FileMode.Open))
        {
            // Create and run job
            // Note that there are methods available for formats other than CSV
            BigQueryJob job = client.UploadCsv(
                datasetId, tableId, null, stream, uploadCsvOptions);
            job = job.PollUntilCompleted().ThrowOnAnyError();  // Waits for the job to complete.

            // Display the number of rows uploaded
            BigQueryTable table = client.GetTable(datasetId, tableId);
            Console.WriteLine(
                $"Loaded {table.Resource.NumRows} rows to {table.FullyQualifiedId}");
        }
    }
}

import (
	"context"
	"fmt"
	"os"

	"cloud.google.com/go/bigquery"
)

// importCSVFromFile demonstrates loading data into a BigQuery table using a file on the local filesystem.
func importCSVFromFile(projectID, datasetID, tableID, filename string) error {
	// projectID := "my-project-id"
	// datasetID := "mydataset"
	// tableID := "mytable"
	ctx := context.Background()
	client, err := bigquery.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("bigquery.NewClient: %v", err)
	}
	defer client.Close()

	f, err := os.Open(filename)
	if err != nil {
		return err
	}
	source := bigquery.NewReaderSource(f)
	source.AutoDetect = true   // Allow BigQuery to determine schema.
	source.SkipLeadingRows = 1 // CSV has a single header line.

	loader := client.Dataset(datasetID).Table(tableID).LoaderFrom(source)

	job, err := loader.Run(ctx)
	if err != nil {
		return err
	}
	status, err := job.Wait(ctx)
	if err != nil {
		return err
	}
	if err := status.Err(); err != nil {
		return err
	}
	return nil
}

TableId tableId = TableId.of(datasetName, tableName);
WriteChannelConfiguration writeChannelConfiguration =
    WriteChannelConfiguration.newBuilder(tableId).setFormatOptions(FormatOptions.csv()).build();
// The location must be specified; other fields can be auto-detected.
JobId jobId = JobId.newBuilder().setLocation(location).build();
TableDataWriteChannel writer = bigquery.writer(jobId, writeChannelConfiguration);
// Write data to writer
try (OutputStream stream = Channels.newOutputStream(writer)) {
  Files.copy(csvPath, stream);
}
// Get load job
Job job = writer.getJob();
job = job.waitFor();
LoadStatistics stats = job.getStatistics();
return stats.getOutputRows();

// Imports the Google Cloud client library
const {BigQuery} = require('@google-cloud/bigquery');
const bigquery = new BigQuery();

async function loadLocalFile() {
  // Imports a local file into a table.

  /**
   * TODO(developer): Uncomment the following lines before running the sample.
   */
  // const filename = '/path/to/file.csv';
  // const datasetId = 'my_dataset';
  // const tableId = 'my_table';

  // Load data from a local file into the table
  const [job] = await bigquery
    .dataset(datasetId)
    .table(tableId)
    .load(filename);

  console.log(`Job ${job.id} completed.`);

  // Check the job's status for errors
  const errors = job.status.errors;
  if (errors && errors.length > 0) {
    throw errors;
  }
}

use Google\Cloud\BigQuery\BigQueryClient;
use Google\Cloud\Core\ExponentialBackoff;

/** Uncomment and populate these variables in your code */
// $projectId  = 'The Google project ID';
// $datasetId  = 'The BigQuery dataset ID';
// $tableId    = 'The BigQuery table ID';
// $source     = 'The path to the CSV source file to import';

// instantiate the bigquery table service
$bigQuery = new BigQueryClient([
    'projectId' => $projectId,
]);
$dataset = $bigQuery->dataset($datasetId);
$table = $dataset->table($tableId);
// create the import job
$loadConfig = $table->load(fopen($source, 'r'))->sourceFormat('CSV');

$job = $table->runJob($loadConfig);
// poll the job until it is complete
$backoff = new ExponentialBackoff(10);
$backoff->execute(function () use ($job) {
    printf('Waiting for job to complete' . PHP_EOL);
    $job->reload();
    if (!$job->isComplete()) {
        throw new Exception('Job has not yet completed', 500);
    }
});
// check if the job has errors
if (isset($job->info()['status']['errorResult'])) {
    $error = $job->info()['status']['errorResult']['message'];
    printf('Error running job: %s' . PHP_EOL, $error);
} else {
    print('Data imported successfully' . PHP_EOL);
}

from google.cloud import bigquery

# Construct a BigQuery client object.
client = bigquery.Client()

# TODO(developer): Set table_id to the ID of the table to create.
# table_id = "your-project.your_dataset.your_table_name"

job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.CSV, skip_leading_rows=1, autodetect=True,
)

with open(file_path, "rb") as source_file:
    job = client.load_table_from_file(source_file, table_id, job_config=job_config)

job.result()  # Waits for the job to complete.

table = client.get_table(table_id)  # Make an API request.
print(
    "Loaded {} rows and {} columns to {}".format(
        table.num_rows, len(table.schema), table_id
    )
)

require "google/cloud/bigquery"

def load_from_file dataset_id = "your_dataset_id",
                   file_path  = "path/to/file.csv"
  bigquery = Google::Cloud::Bigquery.new
  dataset  = bigquery.dataset dataset_id
  table_id = "new_table_id"

  # Infer the config.location based on the location of the referenced dataset.
  load_job = dataset.load_job table_id, file_path do |config|
    config.skip_leading = 1
    config.autodetect   = true
  end
  load_job.wait_until_done! # Waits for table load to complete.

  table = dataset.table table_id
  puts "Loaded #{table.rows_count} rows into #{table.id}"
end

制限事項

ローカル データソースからのデータの読み込みには、次の制限があります。

• ワイルドカードやカンマ区切りのリストは、ローカル データソースからファイルを読み込む場合にはサポートされません。ファイルは個別に読み込む必要があります。
• Google Cloud コンソールを使用する場合、ローカル データソースから読み込まれるファイルのサイズが 100 MB を超えないようにしてください。サイズの大きいファイルの場合は、Cloud Storage からファイルを読み込みます。

ジョブの読み込み容量

クエリのオンデマンド モードと同様に、読み込みジョブはデフォルトでスロットの共有プールを使用します。BigQuery では、この共有プールの使用可能容量や読み込みジョブのスループットが保証されません。

スループットを増やすか、読み込みジョブの容量を予測どおりに制御するには、スロット予約を作成し、読み込みジョブの実行専用の PIPELINE スロットを割り当てます。詳細については、予約割り当てをご覧ください。

圧縮データと非圧縮データを読み込む

Avro、Parquet、ORC の形式の場合、BigQuery では、サポートされているコーデックを使用してファイルデータが圧縮されているファイルの読み込みがサポートされています。ただし BigQuery では、gzip ユーティリティなどを使用して圧縮されたこれらの形式のファイルの読み込みはサポートされていません。

Avro バイナリ形式は、圧縮データと非圧縮データの両方の読み込みに適しています。データブロックが圧縮されていても、Avro データは並行して読み取ることができるため、より高速に読み込みが行われます。サポートされている圧縮コーデックの一覧については、Avro 圧縮をご覧ください。

Parquet の効率的な列単位のエンコードでは通常、圧縮率が高くなり、ファイルサイズが小さくなるため、Parquet バイナリ形式も適しています。Parquet ファイルは、ファイルの並列読み込みが可能な圧縮手法も使用します。サポートされている圧縮コーデックの一覧については、Parquet 圧縮をご覧ください。

ORC バイナリ形式には、Parquet 形式に似た利点があります。データ ストライプを並列で読み取ることができるため、ORC ファイルのデータは高速で読み取ることができます。各データ ストライプの行が順番に読み込まれます。読み込み時間を短縮するには、データ ストライプのサイズを 256 MB 以下にします。サポートされている圧縮コーデックの一覧については、ORC 圧縮をご覧ください。

CSV や JSON などのデータ形式の場合、BigQuery では非圧縮ファイルを並列読み取りできるため、非圧縮ファイルを圧縮ファイルよりかなり速く読み込むことができます。ただし、非圧縮ファイルはサイズが大きいため、使用すると帯域幅の制限に達し、BigQuery に読み込まれる前に Cloud Storage にステージングされるデータのために Cloud Storage のコストが高くなる可能性があります。圧縮ファイルや非圧縮ファイルの場合、行の順序が保証されないことにもご注意ください。使用状況に応じて、どちらを重視するかを決定する必要があります。

一般に、帯域幅が限られている場合は、CSV ファイルや JSON ファイルを gzip で圧縮してから Cloud Storage にアップロードします。BigQuery にデータを読み込むときに CSV ファイルと JSON ファイルでサポートされているファイル圧縮形式は gzip のみです。読み込み速度が重要なアプリを使用していて、データの読み込み用に豊富な帯域幅がある場合は、ファイルを圧縮しないでおきます。

テーブルへの追加または上書き

テーブルに追加のデータを読み込むには、ソースファイルを使用するか、クエリ結果を追加します。データのスキーマが追加先のテーブルまたはパーティションのスキーマと一致しない場合は、追加または上書きするときにスキーマを更新できます。

データの追加時にスキーマを更新する場合、BigQuery では次のことが可能です。

• 新しいフィールドを追加する
• フィールドのモードを REQUIRED から NULLABLE に緩和する

テーブルを上書きする場合、スキーマは必ず上書きされます。テーブルを上書きするとき、スキーマの更新は制限されません。

Google Cloud コンソールでは、[書き込み設定] オプションを使用して、ソースファイルやクエリ結果からデータを読み込むときに行う操作を指定します。bq コマンドライン ツールと API には、次のオプションがあります。

割り当てポリシー

データのバッチ読み込みの割り当てポリシーについては、割り当てと上限のページの読み込みジョブをご覧ください。

現在の割り当て使用量を確認する

クエリ、読み込み、抽出、コピージョブの現在の使用状況を確認するには、INFORMATION_SCHEMA クエリを実行して、指定した期間に実行されたジョブに関するメタデータを表示します。現在の使用量と割り当て上限を比較することで、特定の種類のジョブの割り当て使用量を確認できます。次のクエリの例では、INFORMATION_SCHEMA.JOBS ビューを使用して、プロジェクトごとにクエリ、読み込み、抽出、コピーの各ジョブの数を一覧取得します。

SELECT
  sum(case  when job_type="QUERY" then 1 else 0 end) as QRY_CNT,
  sum(case  when job_type="LOAD" then 1 else 0 end) as LOAD_CNT,
  sum(case  when job_type="EXTRACT" then 1 else 0 end) as EXT_CNT,
  sum(case  when job_type="COPY" then 1 else 0 end) as CPY_CNT
FROM `region-eu`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE date(creation_time)= CURRENT_DATE()

料金

共有スロットプールを使用して BigQuery にデータを一括で読み込む場合、料金は発生しません。詳しくは、BigQuery データ取り込みの料金をご覧ください。

使用例

決められた期限までに完了する必要がある夜間のバッチ処理パイプラインがあるとします。規制当局に送信するレポートを生成するには、別のバッチ処理でさらに処理を行ってこの期限までにデータを使用可能にする必要があります。このユースケースは金融などの規制が多い業界では一般的です。

期限が満たされていれば、レイテンシは問題にならないため、このユースケースに適したアプローチは、読み込みジョブを使用したデータの一括読み込みとなります。Cloud Storage バケットが BigQuery データセットにデータを読み込むためのロケーションの要件を満たしていることを確認します。

BigQuery の読み込みジョブの結果はアトミックです。すべてのレコードが挿入されるか、どれも挿入されません。ベスト プラクティスとして、単一の読み込みジョブにすべてのデータを挿入する場合は、JobConfigurationLoad リソースの WRITE_TRUNCATE 処理を使用して新しいテーブルを作成します。たとえば、成功状態をクライアントに返す際など、クライアントが失敗したジョブと引き起こされた失敗を区別できない可能性があるため、これは失敗した読み込みジョブを再試行する場合に重要です。

取り込まれるデータがすでに Cloud Storage に正常にコピーされていると仮定すれば、指数バックオフを使用して再試行するだけで取り込みの失敗に対処できます。

再試行を含めても、夜間のバッチジョブでテーブルごとに 1 日あたり 1,500 回の読み込みというデフォルトの割り当てに達しないようにすることをおすすめします。データを段階的に読み込む場合は、デフォルトの割り当てで 5 分ごとに読み込みジョブを実行できますが、ジョブごとに少なくとも 1 回の再試行の未消化の割り当てが必要です。