Storage Write API（REST）を使用する

このドキュメントでは、以前は以前の tabledata.insertAll メソッドと呼ばれていた BigQuery Storage Write API（REST）を使用して、BigQuery にデータをストリーミングする方法について説明します。

新しいプロジェクトでは、Storage Write API（REST）ではなく BigQuery Storage Write API（gRPC）を使用することをおすすめします。Storage Write API（gRPC）は、1 回限りの配信セマンティクスや Apache Iceberg マネージド テーブルへのストリーミングなど、低価格かつ堅牢な機能を備えています。既存のプロジェクトを Storage Write API（REST）から Storage Write API（gRPC）に移行する場合は、デフォルト ストリームを選択することをおすすめします。Storage Write API（REST）は引き続き完全にサポートされます。

始める前に

1. 宛先テーブルを含んだデータセットへの書き込みアクセス権があることを確認します。テーブルにデータを書き込む前に、まずテーブルが存在している必要があります。ただし、テンプレート テーブルを使用する場合は異なります。テンプレート テーブルについて詳しくは、テンプレート テーブルを使用した自動的なテーブル作成をご覧ください。

2. データ ストリーミングの割り当てポリシーを確認します。

3. Google Cloud プロジェクトに対して課金が有効になっていることを確認します。

無料枠でストリーミングは利用できません。課金を有効にせずストリーミングの使用を試みると、次のエラーが表示されます。BigQuery: Streaming insert is not allowed in the free tier.

4. このドキュメントの各タスクを実行するために必要な権限をユーザーに与える Identity and Access Management（IAM）のロールを付与します。

必要な権限

BigQuery にデータをストリーミングするには、次の IAM 権限が必要です。

• bigquery.tables.updateData（テーブルにデータを挿入）
• bigquery.tables.get（テーブルのメタデータを取得）
• bigquery.datasets.get（データセットのメタデータを取得）
• bigquery.tables.create（テンプレート テーブルを使用してテーブルを自動的に作成する場合は必須）

次の各 IAM 事前定義ロールには、BigQuery にデータをストリーミングするために必要な権限が含まれています。

• roles/bigquery.dataEditor
• roles/bigquery.dataOwner
• roles/bigquery.admin

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

BigQuery にデータをストリーミングする

using Google.Cloud.BigQuery.V2;

public class BigQueryTableInsertRows
{
    public void TableInsertRows(
        string projectId = "your-project-id",
        string datasetId = "your_dataset_id",
        string tableId = "your_table_id"
    )
    {
        BigQueryClient client = BigQueryClient.Create(projectId);
        BigQueryInsertRow[] rows = new BigQueryInsertRow[]
        {
            // The insert ID is optional, but can avoid duplicate data
            // when retrying inserts.
            new BigQueryInsertRow(insertId: "row1") {
                { "name", "Washington" },
                { "post_abbr", "WA" }
            },
            new BigQueryInsertRow(insertId: "row2") {
                { "name", "Colorado" },
                { "post_abbr", "CO" }
            }
        };
        client.InsertRows(datasetId, tableId, rows);
    }
}

import (
	"context"
	"fmt"

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

// Item represents a row item.
type Item struct {
	Name string
	Age  int
}

// Save implements the ValueSaver interface.
// This example disables best-effort de-duplication, which allows for higher throughput.
func (i *Item) Save() (map[string]bigquery.Value, string, error) {
	return map[string]bigquery.Value{
		"full_name": i.Name,
		"age":       i.Age,
	}, bigquery.NoDedupeID, nil
}

// insertRows demonstrates inserting data into a table using the streaming insert mechanism.
func insertRows(projectID, datasetID, tableID 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: %w", err)
	}
	defer client.Close()

	inserter := client.Dataset(datasetID).Table(tableID).Inserter()
	items := []*Item{
		// Item implements the ValueSaver interface.
		{Name: "Phred Phlyntstone", Age: 32},
		{Name: "Wylma Phlyntstone", Age: 29},
	}
	if err := inserter.Put(ctx, items); err != nil {
		return err
	}
	return nil
}

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryError;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.InsertAllRequest;
import com.google.cloud.bigquery.InsertAllResponse;
import com.google.cloud.bigquery.TableId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Sample to inserting rows into a table without running a load job.
public class TableInsertRows {

  public static void main(String[] args) {
    // TODO(developer): Replace these variables before running the sample.
    String datasetName = "MY_DATASET_NAME";
    String tableName = "MY_TABLE_NAME";
    // Create a row to insert
    Map<String, Object> rowContent = new HashMap<>();
    rowContent.put("booleanField", true);
    rowContent.put("numericField", "3.14");
    // TODO(developer): Replace the row id with a unique value for each row.
    String rowId = "ROW_ID";
    tableInsertRows(datasetName, tableName, rowId, rowContent);
  }

  public static void tableInsertRows(
      String datasetName, String tableName, String rowId, Map<String, Object> rowContent) {
    try {
      // Initialize client that will be used to send requests. This client only needs to be created
      // once, and can be reused for multiple requests.
      BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

      // Get table
      TableId tableId = TableId.of(datasetName, tableName);

      // Inserts rowContent into datasetName:tableId.
      InsertAllResponse response =
          bigquery.insertAll(
              InsertAllRequest.newBuilder(tableId)
                  // More rows can be added in the same RPC by invoking .addRow() on the builder.
                  // You can omit the unique row ids to disable de-duplication.
                  .addRow(rowId, rowContent)
                  .build());

      if (response.hasErrors()) {
        // If any of the insertions failed, this lets you inspect the errors
        for (Map.Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) {
          System.out.println("Response error: \n" + entry.getValue());
        }
      }
      System.out.println("Rows successfully inserted into table");
    } catch (BigQueryException e) {
      System.out.println("Insert operation not performed \n" + e.toString());
    }
  }
}

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

async function insertRowsAsStream() {
  // Inserts the JSON objects into my_dataset:my_table.

  /**
   * TODO(developer): Uncomment the following lines before running the sample.
   */
  // const datasetId = 'my_dataset';
  // const tableId = 'my_table';
  const rows = [
    {name: 'Tom', age: 30},
    {name: 'Jane', age: 32},
  ];

  // Insert data into a table
  await bigquery.dataset(datasetId).table(tableId).insert(rows);
  console.log(`Inserted ${rows.length} rows`);
}

use Google\Cloud\BigQuery\BigQueryClient;

/**
 * Stream data into bigquery
 *
 * @param string $projectId The project Id of your Google Cloud Project.
 * @param string $datasetId The BigQuery dataset ID.
 * @param string $tableId The BigQuery table ID.
 * @param string $data Json encoded data For eg,
 *    $data = json_encode([
 *       "field1" => "value1",
 *       "field2" => "value2",
 *    ]);
 */
function stream_row(
    string $projectId,
    string $datasetId,
    string $tableId,
    string $data
): void {
    // instantiate the bigquery table service
    $bigQuery = new BigQueryClient([
      'projectId' => $projectId,
    ]);
    $dataset = $bigQuery->dataset($datasetId);
    $table = $dataset->table($tableId);

    $data = json_decode($data, true);
    $insertResponse = $table->insertRows([
      ['data' => $data],
      // additional rows can go here
    ]);
    if ($insertResponse->isSuccessful()) {
        print('Data streamed into BigQuery successfully' . PHP_EOL);
    } else {
        foreach ($insertResponse->failedRows() as $row) {
            foreach ($row['errors'] as $error) {
                printf('%s: %s' . PHP_EOL, $error['reason'], $error['message']);
            }
        }
    }
}

from google.cloud import bigquery

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

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

rows_to_insert = [
    {"full_name": "Phred Phlyntstone", "age": 32},
    {"full_name": "Wylma Phlyntstone", "age": 29},
]

errors = client.insert_rows_json(table_id, rows_to_insert)  # Make an API request.
if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

require "google/cloud/bigquery"

def table_insert_rows dataset_id = "your_dataset_id", table_id = "your_table_id"
  bigquery = Google::Cloud::Bigquery.new
  dataset  = bigquery.dataset dataset_id
  table    = dataset.table table_id

  row_data = [
    { name: "Alice", value: 5  },
    { name: "Bob",   value: 10 }
  ]
  response = table.insert row_data

  if response.success?
    puts "Inserted rows successfully"
  else
    puts "Failed to insert #{response.error_rows.count} rows"
  end
end

行の挿入時に insertID フィールドに値を入力する必要はありません。次の例は、ストリーミング時に各行の insertID の送信を防ぐ方法を示しています。

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryError;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.InsertAllRequest;
import com.google.cloud.bigquery.InsertAllResponse;
import com.google.cloud.bigquery.TableId;
import com.google.common.collect.ImmutableList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Sample to insert rows without row ids in a table
public class TableInsertRowsWithoutRowIds {

  public static void main(String[] args) {
    // TODO(developer): Replace these variables before running the sample.
    String datasetName = "MY_DATASET_NAME";
    String tableName = "MY_TABLE_NAME";
    tableInsertRowsWithoutRowIds(datasetName, tableName);
  }

  public static void tableInsertRowsWithoutRowIds(String datasetName, String tableName) {
    try {
      // Initialize client that will be used to send requests. This client only needs to be created
      // once, and can be reused for multiple requests.
      final BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
      // Create rows to insert
      Map<String, Object> rowContent1 = new HashMap<>();
      rowContent1.put("stringField", "Phred Phlyntstone");
      rowContent1.put("numericField", 32);
      Map<String, Object> rowContent2 = new HashMap<>();
      rowContent2.put("stringField", "Wylma Phlyntstone");
      rowContent2.put("numericField", 29);
      InsertAllResponse response =
          bigquery.insertAll(
              InsertAllRequest.newBuilder(TableId.of(datasetName, tableName))
                  // No row ids disable de-duplication, and also disable the retries in the Java
                  // library.
                  .setRows(
                      ImmutableList.of(
                          InsertAllRequest.RowToInsert.of(rowContent1),
                          InsertAllRequest.RowToInsert.of(rowContent2)))
                  .build());

      if (response.hasErrors()) {
        // If any of the insertions failed, this lets you inspect the errors
        for (Map.Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) {
          System.out.println("Response error: \n" + entry.getValue());
        }
      }
      System.out.println("Rows successfully inserted into table without row ids");
    } catch (BigQueryException e) {
      System.out.println("Insert operation not performed \n" + e.toString());
    }
  }
}

from google.cloud import bigquery

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

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

rows_to_insert = [
    {"full_name": "Phred Phlyntstone", "age": 32},
    {"full_name": "Wylma Phlyntstone", "age": 29},
]

errors = client.insert_rows_json(
    table_id, rows_to_insert, row_ids=[None] * len(rows_to_insert)
)  # Make an API request.
if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

日時データを送信する

日時フィールドの場合、Storage Write API（REST）で次のようにデータをフォーマットします。

範囲データを送信する

RANGE<T> 型のフィールドの場合は、Storage Write API（REST）内のデータの形式を、start と end の 2 つのフィールドを持つ JSON オブジェクトにします。start フィールドと end フィールドに値が指定されていない場合、または値が NULL の場合は、境界が存在しないことを表します。これらのフィールドは、タイプ T のサポートされている JSON 形式（T は DATE、DATETIME、TIMESTAMP のいずれか）と同じである必要があります。

次の例では、f_range_date フィールドはテーブルの RANGE<DATE> 列を表します。Storage Write API（REST）を使用して、この列に 1 行挿入します。

{
    "f_range_date": {
        "start": "1970-01-02",
        "end": null
    }
}

ストリーム データの可用性

GoogleSQL クエリを使用したリアルタイム分析にデータは、BigQuery が Storage Write API（REST）リクエストの確認応答に成功した直後に使用できるようになります。オンデマンド コンピューティングの料金を使用している場合、ストリーミング バッファ内のデータに対してクエリを実行しても、ストリーミング バッファから処理されたバイト数に対して料金は発生しません。容量ベースの料金を使用している場合、予約は、ストリーミング バッファ内のデータを処理するためにスロットを消費します。

最近取り込み時間パーティション分割テーブルにストリーミングした行は、_PARTITIONTIME 疑似列の値が一時的に NULL になります。このような行に対して、BigQuery はバックグラウンドで PARTITIONTIME 列の最終的な NULL 以外の値を割り当てます。これは通常、数分以内に行われます。まれに最長 90 分かかることがあります。

最近ストリーミングされた行の一部は、通常、数分間はテーブルのコピーに使用できない可能性があります。まれに最長 90 分かかることがあります。データがテーブルのコピーに使用できるかどうかを確認するには、tables.get に対するレスポンスで streamingBuffer というセクションをチェックします。streamingBuffer セクションがなければ、データはコピーに利用できます。streamingBuffer.oldestEntryTime フィールドを使用して、ストリーミング バッファ内のレコードの経過時間を特定することもできます。

ベスト エフォート型の重複排除

挿入された行に対して insertId を指定した場合、BigQuery はこの ID を使用して、ベスト エフォート型の重複排除を最大 1 分間サポートします。つまり、その期間内に同じ insertId の同じ行を同じテーブルに複数回ストリーミングしようとすると、BigQuery はその行の複数のオカレンスを重複排除して、それらのオカレンスの一つだけを保持する可能性があります。

このシステムでは、同じ insertId が指定された行も同一のものであると想定されます。2 つの行が同じ insertId を持つ場合、どちらの行を BigQuery が保持するかは非決定的になります。

重複排除は一般に、システムと BigQuery 間のネットワーク エラーや BigQuery の内部エラーといった特定のエラー状態でストリーミング挿入の状態を判断する方法がない分散システムでの再試行シナリオ向けです。挿入を再試行する場合は、BigQuery がデータの重複排除を試行できるよう、同じ行セットに同じ insertId を使用するようにしてください。詳細については、ストリーミング挿入に関するトラブルシューティングをご覧ください。

BigQuery の重複排除はベスト エフォート型であり、データの重複がないことを保証するメカニズムとしての使用には適していません。さらに、データの高い信頼性と可用性を保証するために、BigQuery はベスト エフォート型の重複排除の品質を低下させる可能性があります。

データの重複排除に関して厳密な要件がある場合は、Google Cloud Datastore がトランザクションをサポートする代替サービスとなります。

ベスト エフォート型の重複排除の無効化

ベスト エフォート型の重複排除を無効にするには、挿入された各行の insertId フィールドに値を設定しないようにします。これは、データの挿入を行う場合に推奨される方法です。

Apache Beam と Dataflow

Apache Beam の Java 用 BigQuery I/O コネクタを使用しているときにベスト エフォート型の重複排除を無効にするには、ignoreInsertIds() メソッドを使用します。

手動の重複排除

ストリーミングの実行後に重複行が残らないようにするには、次の手動プロセスを使用します。

1. テーブル スキーマ内の列として insertId を追加し、各行のデータに insertId 値を含めます。
2. ストリーミングが停止した後に、次のクエリを実行して重複をチェックします。

   #standardSQL
   SELECT
     MAX(count) FROM(
     SELECT
       ID_COLUMN,
       count(*) as count
     FROM
       `TABLE_NAME`
     GROUP BY
       ID_COLUMN)

   結果が 1 より大きい場合は、重複が存在します。
3. 重複を排除するには、次のクエリを実行します。宛先テーブルを指定してサイズの大きい結果を許容し、結果のフラット化を無効にします。

   #standardSQL
   SELECT
     * EXCEPT(row_number)
   FROM (
     SELECT
       *,
       ROW_NUMBER()
             OVER (PARTITION BY ID_COLUMN) row_number
     FROM
       `TABLE_NAME`)
   WHERE
     row_number = 1

重複排除クエリに関する注記:

• 重複排除クエリの戦略として、新規テーブルをターゲットにした方が安全です。なお、書き込み処理 WRITE_TRUNCATE を使用して、ソーステーブルをターゲットにすることもできます。
• 重複排除クエリは、値が 1 の row_number 列をテーブル スキーマの末尾に追加します。クエリでは、GoogleSQL の SELECT * EXCEPT ステートメントを使用して、この row_number 列を宛先テーブルから除外します。#standardSQL 接頭辞を使用すると、このクエリで GoogleSQL が有効になります。また、特定の列名を指定して、この列を除外することもできます。
• ライブデータのクエリで重複を排除するには、重複排除クエリを使用してテーブルのビューを作成するという方法もあります。なお、ビューに対するクエリコストはビュー内の選択列に基づいて計算されるため、スキャンされるバイトサイズが大きくなる可能性があります。

時間パーティション分割テーブルへのストリーミング

時間パーティション分割テーブルにデータをストリーミングする場合は、各パーティションにストリーミング バッファが存在します。writeDisposition プロパティを WRITE_TRUNCATE に設定すると、パーティションを上書きする読み込み、クエリ、またはコピーのジョブを実行したときにストリーミング バッファが保持されます。ストリーミング バッファを削除するには、そのパーティションに対して tables.get を呼び出し、ストリーミング バッファが空であることを確認します。

取り込み時間パーティショニング

取り込み時間パーティション分割テーブルにストリーミングすると、BigQuery は現在の UTC 時間から宛先パーティションを推測します。

新しく到着したデータは、ストリーミング バッファ内にある間、一時的に __UNPARTITIONED__ パーティションに配置されます。パーティション分割されていないデータが十分蓄積されると、BigQuery はデータを正しいパーティションに分割します。ただし、データが __UNPARTITIONED__ パーティションから移動するために要する時間に関する SLA はありません。いずれかの疑似列（優先するデータ型に応じて _PARTITIONTIME または _PARTITIONDATE）を使用して、__UNPARTITIONED__ パーティションの NULL 値を除外することにより、ストリーミング バッファ内のデータをクエリの対象外にできます。

日次パーティション分割テーブルにデータをストリーミングする場合は、Storage Write API（REST）の一部としてパーティション デコレータを指定することで、日付の推定をオーバーライドできます。tableId パラメータにデコレータを含めます。たとえば、次のようにパーティション デコレータを使用して、table1 テーブルの 2021-03-01 に対応するパーティションにストリーミングできます。

table1$20210301

パーティション デコレータを使用してストリーミングを行う際は、現在の UTC 時間に基づき、過去 31 日以内のパーティションと現在の日付から 16 日後までのパーティションにストリーミングできます。この範囲に含まれない日付のパーティションに書き込むには、パーティション分割テーブルデータの追加と上書きで説明するように、読み込みジョブまたはクエリジョブを使用します。

パーティション デコレータを使用したストリーミングは、日単位のパーティション分割テーブルでのみサポートされています。時間単位、月単位、年単位のパーティション分割テーブルではサポートされていません。

テストには、 コマンドライン ツールの bq insert CLI コマンドを使用できます。たとえば、次のコマンドを実行すると、2017 年 1 月 1 日（$20170101）のパーティション全体のデータを、mydataset.mytable という名前のパーティション分割テーブルに 1 行がストリーミングされます。

echo '{"a":1, "b":2}' | bq insert 'mydataset.mytable$20170101'

時間単位列パーティショニング

DATE 列、DATETIME 列、または TIMESTAMP 列で分割されたテーブルに、過去 10 年間、向こう 1 年間のデータをストリーミングできます。この範囲外のデータは拒否されます。

データがストリーミングされると、最初に __UNPARTITIONED__ パーティションに配置されます。パーティション分割されていないデータが十分蓄積されると、BigQuery はデータのパーティション再設定を自動的に行い、適切なパーティションに配置します。ただし、データが __UNPARTITIONED__ パーティションから移動するために要する時間に関する SLA はありません。

• 注: 日単位のパーティションは、時間、月、年単位のパーティションとは異なる方法で処理されます。期間（過去 7 日間から将来の 3 日間まで）外のデータのみがパーティション分割されていないパーティションに抽出され、パーティションの再分割を待ちます。一方、時間単位のパーティション分割テーブルの場合、データは常にパーティション分割されていないパーティションに抽出され、後でパーティションの再分割が行われます。

テンプレート テーブルを使用してテーブルを自動的に作成する

テンプレート テーブルは、論理テーブルを多数の小さなテーブルに分割して、より小さなデータの集合（たとえば、ユーザー ID ごと）を作成するメカニズムを提供します。テンプレート テーブルには、以下に説明するいくつかの制限があります。代わりにパーティション分割テーブルとクラスタ化テーブルを使用してこの動作を実現することをおすすめします。

BigQuery API を介してテンプレート テーブルを使用するには、Storage Write API（REST）リクエストに templateSuffix パラメータを追加します。bq コマンドライン ツールの場合は、insert コマンドに template_suffix フラグを追加します。BigQuery は、templateSuffix パラメータまたは template_suffix フラグを検出した場合、ターゲット テーブルをベース テンプレートとして扱います。さらに、ターゲット テーブルと同じスキーマを共有する新しいテーブルを、指定されたサフィックスを含む名前で作成します。

<targeted_table_name> + <templateSuffix>

テンプレート テーブルを使用すると、各テーブルを個別に作成し、各テーブルのスキーマを指定するためのオーバーヘッドを回避できます。テンプレートを 1 つ作成し、複数のサフィックスを指定するだけで、新規テーブルを BigQuery に自動作成させることができます。BigQuery は、各テーブルを同じプロジェクトとデータセット内に配置します。

テンプレート テーブルを通じて作成されたテーブルは、通常数秒以内に利用可能になります。ただし、それより長い時間を要する場合もまれにあります。

テンプレート テーブル スキーマを変更する

テンプレート テーブル スキーマを変更した場合は、それ以降に生成されるすべてのテーブルで、更新後のスキーマが使用されます。以前に生成されたテーブルには影響はありません（既存のテーブルにストリーミング バッファが残っている場合を除く）。

ストリーミング バッファが残っている既存のテーブルについては、テンプレート テーブル スキーマに対する変更が後方互換性のあるものであれば、現にストリーミングが行われているそれらの生成済テーブルのスキーマも更新されます。ただし、テンプレート テーブル スキーマに対する変更が後方互換性のないものである場合は、古いスキーマを使用するバッファデータがすべて失われます。また、互換性がなくなった古いスキーマを使用する既存の生成済テーブルに、新しいデータをストリーミングすることはできません。

テンプレート テーブル スキーマを変更した後は、その変更が伝播されるまで、新しいデータの挿入や、生成されたテーブルに対するクエリを行わないでください。新しいフィールドを挿入するリクエストは、数分以内に処理されます。新規フィールドに対するクエリ実行には、最大 90 分の待機時間を要する場合があります。

生成されたテーブルのスキーマを変更する場合は、テンプレート テーブルを経由したストリーミングが停止し、生成済みテーブルのストリーミング統計セクションが tables.get() レスポンスからなくなる（テーブルでバッファされているデータがなくなる）まで、スキーマを変更しないようにしてください。

パーティション分割テーブルとクラスタ化テーブルには前述の制限がないため、おすすめのメカニズムです。

テンプレート テーブルの詳細

テンプレート サフィックス値

templateSuffix（または --template_suffix）値には、英字（a～z、A～Z）、数字（0～9）、アンダースコア（_）のみを含める必要があります。テーブル名とテーブル サフィックスの最大連結文字数は 1,024 文字です。

割り当て

テンプレート テーブルには、ストリーミング割り当ての制限が適用されます。tables.insert API と同様に、プロジェクトでは、テンプレート テーブルを使用して、1 秒あたり最大 10 個のテーブルを作成できます。この割り当ては作成されるテーブルにのみ適用され、変更されるテーブルには適用されません。

アプリケーションで 1 秒あたり 10 個を超えるテーブルを作成する必要がある場合は、クラスタ化テーブルの使用をおすすめします。たとえば、カーディナリティが高いテーブル ID を単一のクラスタリング テーブルのキー列に配置できます。

有効期間

生成されたテーブルの有効期間はデータセットから継承されます。通常のストリーミング データと同様、生成されたテーブルをすぐにコピーすることはできません。

重複排除

重複排除は、宛先テーブルへの同質な参照間でのみ実行されます。たとえば、テンプレート テーブルと通常の Storage Write API（REST）コマンドの両方を使用して生成済テーブルへのストリーミングを同時に実行した場合、テンプレート テーブルと通常の Storage Write API（REST）コマンドによって挿入された各行の間で重複排除は実行されません。

ビュー

ビューをテンプレート テーブルとして使用することはできません。また、テンプレート テーブルを元にビューを自動生成させることもできません。

ストリーミング挿入のトラブルシューティング

以降のセクションでは、Storage Write API（REST）を使用して BigQuery にデータをストリーミングする際に発生するエラーのトラブルシューティングについて説明します。ストリーミング挿入の割り当てエラーを解決する詳しい方法については、ストリーミング挿入の割り当てエラーをご覧ください。

失敗 HTTP レスポンス コード

ネットワーク エラーなどの失敗 HTTP レスポンス コードが返された場合、ストリーミング挿入が成功したかどうかを確認する方法はありません。リクエストを再送信しようとすると、テーブル内に重複行が発生する可能性があります。テーブルを重複から保護するには、リクエストの送信時に insertId プロパティを設定します。insertId プロパティは BigQuery によって重複排除に使用されます。

権限エラー、無効なテーブル名エラー、または割り当て超過エラーが返された場合、行は挿入されず、リクエスト全体が失敗します。

成功 HTTP レスポンス コード

成功 HTTP レスポンス コードが返された場合でも、BigQuery による行の挿入が部分的にしか成功しなかった可能性があります。レスポンスの insertErrors プロパティをチェックして、行の挿入が成功したかどうかを確認する必要があります。次のシナリオのいずれかが発生することがあります。

• すべての行が正常に挿入されている。insertErrors プロパティが空のリストになっている場合は、すべての行が正常に挿入されています。
• 一部の行が正常に挿入されている。いずれかの行にスキーマの不一致がある場合を除き、insertErrors プロパティで示された行が挿入されておらず、それ以外の行はすべて正常に挿入されています。errors プロパティには、行の挿入が失敗した理由に関する詳細情報が含まれます。index プロパティは、エラーに該当するリクエストの 0 ベースの行インデックスを示します。
• 挿入された行がない。BigQuery がリクエスト内で個別行のスキーマ不一致を検出した場合は、いずれの行も挿入されず、スキーマ不一致がなかった行も含めて、各行に対して insertErrors エントリが返されます。スキーマ不一致がなかった行のエラーは reason プロパティが stopped に設定され、そのまま再送信できます。失敗した行には、スキーマ不一致に関する詳細情報が含まれます。BigQuery データ型でサポートされているプロトコル バッファの型については、サポートされているプロトコル バッファと Arrow のデータ型をご覧ください。

ストリーミング挿入のメタデータ エラー

BigQuery のストリーミング API はインサート率が高くなるように設計されているため、ストリーミング システムを操作する際、基盤となるテーブルのメタデータの変更は結果整合が保たれます。ほとんどの場合、メタデータの変更は数分以内にプロパゲートされますが、その間は API レスポンスで、対象テーブルの不整合な状態が反映されることがあります。

次のようなシナリオがあります。

• スキーマの変更。スキーマの変更がストリーミング システムにすぐに反映されない場合があるため、ストリーミング挿入を最近受け取ったテーブルのスキーマを変更すると、スキーマの不一致エラーが発生することがあります。
• テーブルの作成 / 削除。存在しないテーブルへのストリーミングによって、notFound レスポンスのバリエーションが返されます。レスポンスで作成されたテーブルは、後続のストリーミング挿入ですぐには認識されない可能性があります。同様に、テーブルを削除または、再作成すると、古いテーブルにストリーミング挿入が実質的に配信される期間が生じることがあります。このストリーミング挿入は新しいテーブルには存在しない可能性があります。
• テーブルの切り捨て - （WRITE_TRUNCATE の writeDisposition を使用するクエリジョブによって）テーブルのデータを切り捨てる場合も同様に、整合性期間中の後続の挿入が破棄されることがあります。

欠損 / 利用不可能なデータ

ストリーミング挿入は、書き込み用に最適化されたストレージに一時的に存在します。このストレージは、マネージド ストレージと可用性の特性が異なります。BigQuery の一部のオペレーション（テーブル コピー ジョブや tabledata.list などの API メソッドなど）では、書き込み用に最適化されたストレージは操作されず、最新のストリーミング データが宛先テーブルまたは出力に存在しません。

ストリーミング挿入の割り当てエラー

このセクションでは、BigQuery へのデータのストリーミングに関連する割り当てエラーを解決するためのヒントを紹介します。

特定のリージョンでは、各行の insertId フィールドにデータを入力しないと、ストリーミング挿入の割り当て量が多くなります。ストリーミング挿入の割り当ての詳細については、ストリーミング挿入をご覧ください。BigQuery ストリーミングの割り当て関連エラーは、insertId の有無によって異なります。

エラー メッセージ

insertId フィールドが空の場合、次の割り当てエラーが発生する可能性があります。

insertId フィールドに値が入力されている場合、次の割り当てエラーが発生する可能性があります。

insertId フィールドの目的は、挿入した行の重複を排除することです。数分で同じ insertId が複数挿入されると、BigQuery は 1 つのバージョンのレコードを書き込みます。ただし、この自動重複排除は保証されていません。ストリーミングのスループットを最大にするために、insertId を含めずに手動の重複排除を使用することをおすすめします。詳細については、データ整合性の確保をご覧ください。

このエラーが発生した場合は、問題を診断し、推奨される手順を実施して解決します。

診断

ストリーミング トラフィックの分析には STREAMING_TIMELINE_BY_* ビューを使用します。これは、ストリーミング統計を 1 分間隔で集計し、error_code でグループ化したビューです。割り当てエラーは、結果で error_code が RATE_LIMIT_EXCEEDED または QUOTA_EXCEEDED に等しくなっています。

到達した割り当て上限に従い、total_rows または total_input_bytes を確認します。テーブルレベルの割り当てに関するエラーの場合は、table_id でフィルタリングします。

たとえば次のクエリは、1 分で取り込まれる合計バイト数と割り当てエラーの総数を示します。

SELECT
 start_timestamp,
 error_code,
 SUM(total_input_bytes) as sum_input_bytes,
 SUM(IF(error_code IN ('QUOTA_EXCEEDED', 'RATE_LIMIT_EXCEEDED'),
     total_requests, 0)) AS quota_error
FROM
 `region-REGION_NAME`.INFORMATION_SCHEMA.STREAMING_TIMELINE_BY_PROJECT
WHERE
  start_timestamp > TIMESTAMP_SUB(CURRENT_TIMESTAMP, INTERVAL 1 DAY)
GROUP BY
 start_timestamp,
 error_code
ORDER BY 1 DESC

解決策

この割り当てエラーを解決するには、次の操作を行います。

• ストリーミング割り当ての引き上げをサポートしているリージョンにあるプロジェクトで重複排除に insertId フィールドを使用している場合は、その insertId フィールドの削除をおすすめします。この解決策では、データの重複が発生すると手動で排除しなければならないため、追加の手順が必要になる場合があります。詳細については、手動の重複排除をご覧ください。

• insertId を使用していない場合、または削除できない場合は、24 時間のストリーミング トラフィックをモニタリングし、割り当てエラーを分析します。

  • 発生しているエラーが QUOTA_EXCEEDED ではなく主に RATE_LIMIT_EXCEEDED である場合、トラフィック全体が割り当ての 80% を下回ると、エラーで一時的な急増が示されることがあります。このようなエラーに対処するには、オペレーションを再試行します。その際、次の再試行の前に指数バックオフを行うようにします。

  • Dataflow ジョブを使用してデータを挿入する場合は、ストリーミング挿入ではなく、読み込みジョブの使用を検討してください。詳細については、挿入方法の設定をご覧ください。カスタム I/O コネクタで Dataflow を使用している場合は、代わりに組み込み I/O コネクタの使用を検討してください。詳細については、カスタム I/O パターンをご覧ください。

  • QUOTA_EXCEEDED エラーが発生した場合や、トラフィック全体が割り当ての 80% を常に超えている場合は、割り当ての引き上げをリクエストしてください。詳細については、割り当ての調整をリクエストするをご覧ください。

  • また、ストリーミング挿入を新しい Storage Write API に置き換えることも検討してください。この API は高スループット、低料金であり、多くの便利な機能を使用できます。