您可以选择使用 tabledata.insertAll 方法，以一次一条记录的方式将数据流式插入到 BigQuery 中，而无需通过作业将数据加载到 BigQuery 中。使用此方法可加快数据查询，避免因等待运行加载作业而发生延迟。本文档讨论了在选择此方法之前要权衡的几项重要因素，包括流式插入配额、数据可用性和数据一致性。

准备工作

确保您对包含目标表的数据集具有写入权限。在开始向表中写入数据之前，该表必须存在，除非使用的是模板表。要详细了解模板表，请参阅使用模板表自动创建表。
查看流式插入数据的配额政策。

检查数据可用性

流式插入的数据可用于在首次流式插入表的几秒钟内进行实时分析。在极少数情况下（如中断），流式缓冲区中的数据可能暂时不可用。当数据不可用时，查询仍可继续成功运行，但它们会跳过仍位于流式缓冲区中的某些数据。这些查询将在 bigquery.jobs.getQueryResults 的 errors 字段中（对 bigquery.jobs.query 的响应中）或 bigquery.jobs.get 的 status.errors 字段中包含一个警告。

数据最多可能需要 90 分钟才能用于复制和导出操作。此外，在将数据流式插入到分区表时，流式缓冲区中的数据对应的 _PARTITIONTIME 伪列值为 NULL。要查看数据是否可用于复制和导出，请检查针对 streamingBuffer 部分的 tables.get 响应。如果该部分不存在，则数据应可用于复制或导出，且对应的 _PARTITIONTIME 伪列应具有非 NULL 值。此外，还可利用 streamingBuffer.oldestEntryTime 字段来识别流式缓冲区中记录的存在时间。

确保数据一致性

为帮助确保数据一致性，可为每个插入的行提供 insertId。BigQuery 至少会在一分钟内记住此 ID。如果您尝试在此时间段内流式插入同一组行，并且已设置 insertId 属性，则 BigQuery 将使用 insertId 属性尽可能地删除重复的数据。

您可能必须重试插入，因为在某些错误情况（例如系统和 BigQuery 之间的网络错误或 BigQuery 中的内部错误）下无法确定流式插入的状态。如果重试插入，请对同一组行使用相同的 insertId，让 BigQuery 能够尝试删除重复的数据。如需了解详情，请参阅排查流式插入问题。

在 Google 数据中心意外断开连接的极少数情况下，可能无法自动删除重复的数据。

如果您有更高的数据要求，可选择支持事务的 Google Cloud Datastore 服务。

跨数据位置流式插入数据

您可以将数据同时流式插入到位于美国和欧盟区域的数据集。当 BigQuery 处理 insertAll 请求时，数据可以流经位于数据集位置外部的机器。如果您从数据集位置外部的某一位置流式插入数据，可能会增加发生延迟和错误的几率。

流式插入到按提取时间分区的表

您可以使用 insertAll 请求将各行数据流式插入到分区表。默认情况下，系统会根据当前日期（基于世界协调时间 (UTC)）推断所插入数据对应的目标分区。

如果您要将数据流式插入到按提取时间分区的表，则可以通过在 insertAll 请求中添加分区修饰器来覆盖推断出的日期。例如，您可以使用以下分区修饰器将数据流式插入到表 mydataset.table 中与 2017-03-01 对应的分区：

mydataset.table$20170301

在流式缓冲区中，新传入的数据将暂时关联至 UNPARTITIONED 分区。因此，查询可通过使用某一伪列（[_PARTITIONTIME] 或 [_PARTITIONDATE]，具体取决于您的首选数据类型）过滤掉 UNPARTITIONED 分区中的 NULL 值，将流式缓冲区中的数据从查询中排除。

在使用分区修饰器流式插入数据时，您可以将数据流式插入到日期介于当前日期（基于当前世界协调时间 (UTC)）之前 31 天至之后 16 天之间的分区。要将数据写入到日期在这些允许范围之外的分区，可使用加载作业或查询作业，相关说明请参阅对分区表数据执行附加和覆盖操作。

流式插入到分区表

您可以将数据流式插入到按 DATE 或 TIMESTAMP 列分区且列值介于过去 1 年到未来 6 个月之间的表。不在此范围内的数据会被拒绝。

流式插入数据时，介于过去 7 天到未来 3 天之间的数据会先被放入流式缓冲区中，然后再提取到相应分区。超出此时段的数据（但日期仍然介于过去 1 年到未来 6 个月之间）会先被放入流式缓冲区中，然后再提取到 UNPARTITIONED 分区。如果已有足够多的未分区数据，这些数据会被加载到相应的分区。

使用模板表自动创建表

将数据流式插入到 BigQuery 的常见使用模式是将逻辑表拆分为许多较小的表，以创建较小的数据集（例如按用户 ID）。要按日期创建较小的数据集，请使用分区表。要创建非基于日期的小型表，请使用模板表，然后 BigQuery 会为您创建这些表。

要通过 BigQuery API 使用模板表，请在 insertAll 请求中添加 templateSuffix 参数。对于 bq 命令行工具，请将 template_suffix 标志添加到 insert 命令。如果 BigQuery 检测到 templateSuffix 参数或 template_suffix 标志，它会将目标表视为基本模板并创建一个新表，该表与目标表使用同一架构且其名称包含指定的后缀：

<targeted_table_name> + <templateSuffix>

使用模板表，可免去逐一创建每个表以及为每个表指定架构的开销。您只需创建单个模板并提供不同后缀，BigQuery 便可为您创建新表。BigQuery 会将这些表置于同一项目和数据集中。模板还让架构更新变得更为轻松，因为您只需更新模板表即可。

通过模板表创建的表通常只需几秒即可使用。极少数情况下可能需要更多时间。

更改模板表架构

如果更改模板表架构，则后续生成的所有表都将使用更新的架构。之前生成的表不受影响，除非现有表仍具有流式缓冲区。

对于仍具有流式缓冲区的现有表，如果以向后兼容的方式修改模板表架构，则以主动流式插入方式生成的表的架构也会获得更新。但是，如果不以向后兼容的方式修改模板表架构，则所有使用旧架构的缓冲数据都将丢失。此外，您还无法将新的数据流式插入到已生成的现有表中，这些表使用旧的但已不兼容的架构。

更改模板表架构后，请等待更改传播完成，然后再尝试插入新的数据或查询生成的表。插入新字段的请求几分钟就会成功。而尝试查询新字段最多可能需要等待 90 分钟。

如果要更改已生成的表的架构，除非通过模板表的流式插入已停止，并且 tables.get() 响应中没有已生成表的流式插入统计信息部分（表示表中没有缓冲的数据），否则请勿更改架构。

模板表详情

模板后缀值: templateSuffix（或 --template_suffix）值只能包含字母（a-z、A-Z）、数字 (0-9) 或下划线 (_)。表名称和表后缀总共不得超过 1024 个字符。
配额: 无论是基于模板还是手动创建，所有表均具有同一配额。
生存时间: 生成的表从数据集继承到期时间。与普通的流式插入数据一样，无法立即复制或导出生成的表。
重复信息删除: 仅会在针对目标表的相同引用之间发生重复信息删除。例如，如果使用模板表和常规 insertAll 命令同时将数据流式插入到生成的表，则在通过模板表和常规 insertAll 命令插入的行之间不会发生重复信息删除。
视图: 模板表和生成的表不应为视图。

使用情况示例

大量事件日志

如果您有实时收集大量数据的应用，则流式插入会是一个不错的选择。通常，这类应用具有以下标准：

非事务性。 大量、不断附加行。应用可容忍偶尔出现重复项或数据暂时不可用。
汇总分析。 执行查询通常是为了进行趋势分析，而不会选择单条或小范围记录。

事件跟踪就属于一种大量事件日志的情况。假设您有一个跟踪事件的移动应用。您的应用或移动服务器可以独立记录用户交互或系统错误，并将其流式插入到 BigQuery 中。您可分析此数据以判断整体趋势（例如高度交互或问题方面），并实时监控错误情况。

手动移除重复项

您可以通过下列手动过程，确保在流式插入完成后没有重复的行。

将 insertId 作为列添加到表架构中，并在每行的数据中包含 insertId 值。

在流式插入停止后，执行下列查询以检查重复项：

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

如果结果大于 1，则表示存在重复项。

要移除重复项，请执行以下查询。应指定目标表、允许大型结果并停用结果扁平化功能。

#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 列。该查询使用标准 SQL 中的 SELECT * EXCEPT 语句将 row_number 列从目标表中排除。#standardSQL 前缀用于为此查询启用标准 SQL。或者，您也可按特定列名称进行选择以忽略此列。
要查询移除了重复项的实时数据，您还可以使用重复项移除查询在表上创建视图。请注意，将根据视图中所选的列计算视图的查询费用，这可能导致字节扫描大小较大。

排查流式插入问题

要了解如何在流式插入期间排查错误，请参阅“排查错误”页面上的排查流式插入问题。

返回页首

流式插入示例

C#

在试用此示例之前，请按照《BigQuery 快速入门：使用客户端库》中的 C# 设置说明进行操作。如需了解详情，请参阅 BigQuery C# API 参考文档。

在 GitHub 上查看反馈

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

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

Go

在试用此示例之前，请按照《BigQuery 快速入门：使用客户端库》中的 Go 设置说明进行操作。如需了解详情，请参阅 BigQuery Go API 参考文档。

在 GitHub 上查看反馈

// To run this sample, you will need to create (or reuse) a context and
// an instance of the bigquery client.  For example:
// import "cloud.google.com/go/bigquery"
// ctx := context.Background()
// client, err := bigquery.NewClient(ctx, "your-project-id")
u := client.Dataset(datasetID).Table(tableID).Uploader()
items := []*Item{
	// Item implements the ValueSaver interface.
	{Name: "Phred Phlyntstone", Age: 32},
	{Name: "Wylma Phlyntstone", Age: 29},
}
if err := u.Put(ctx, items); err != nil {
	return err
}

Java

在试用此示例之前，请按照《BigQuery 快速入门：使用客户端库》中的 Java 设置说明进行操作。如需了解详情，请参阅 BigQuery Java API 参考文档。

在 GitHub 上查看反馈

TableId tableId = TableId.of(datasetName, tableName);
// Values of the row to insert
Map<String, Object> rowContent = new HashMap<>();
rowContent.put("booleanField", true);
// Bytes are passed in base64
rowContent.put("bytesField", "Cg0NDg0="); // 0xA, 0xD, 0xD, 0xE, 0xD in base64
// Records are passed as a map
Map<String, Object> recordsContent = new HashMap<>();
recordsContent.put("stringField", "Hello, World!");
rowContent.put("recordField", recordsContent);
InsertAllResponse response =
    bigquery.insertAll(
        InsertAllRequest.newBuilder(tableId)
            .addRow("rowId", rowContent)
            // More rows can be added in the same RPC by invoking .addRow() on the builder
            .build());
if (response.hasErrors()) {
  // If any of the insertions failed, this lets you inspect the errors
  for (Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) {
    // inspect row error
  }
}

Node.js

在试用此示例之前，请按照《BigQuery 快速入门：使用客户端库》中的 Node.js 设置说明进行操作。如需了解详情，请参阅 BigQuery Node.js API 参考文档。

在 GitHub 上查看反馈

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

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

// Creates a client
const bigquery = new BigQuery({
  projectId: projectId,
});

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

PHP

在试用此示例之前，请按照《BigQuery 快速入门：使用客户端库》中的 PHP 设置说明进行操作。如需了解详情，请参阅 BigQuery PHP API 参考文档。

在 GitHub 上查看反馈

use Google\Cloud\BigQuery\BigQueryClient;

/** Uncomment and populate these variables in your code */
// $projectId = 'The Google project ID';
// $datasetId = 'The BigQuery dataset ID';
// $tableId   = 'The BigQuery table ID';
// $data = [
//     "field1" => "value1",
//     "field2" => "value2",
// ];

// instantiate the bigquery table service
$bigQuery = new BigQueryClient([
    'projectId' => $projectId,
]);
$dataset = $bigQuery->dataset($datasetId);
$table = $dataset->table($tableId);

$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']);
        }
    }
}

Python

在试用此示例之前，请按照《BigQuery 快速入门：使用客户端库》中的 Python 设置说明进行操作。如需了解详情，请参阅 BigQuery Python API 参考文档。

在 GitHub 上查看反馈

# TODO(developer): Uncomment the lines below and replace with your values.
# from google.cloud import bigquery
# client = bigquery.Client()
# dataset_id = 'my_dataset'  # replace with your dataset ID
# For this sample, the table must already exist and have a defined schema
# table_id = 'my_table'  # replace with your table ID
# table_ref = client.dataset(dataset_id).table(table_id)
# table = client.get_table(table_ref)  # API request

rows_to_insert = [
    (u'Phred Phlyntstone', 32),
    (u'Wylma Phlyntstone', 29),
]

errors = client.insert_rows(table, rows_to_insert)  # API request

assert errors == []

Ruby

在试用此示例之前，请按照《BigQuery 快速入门：使用客户端库》中的 Ruby 设置说明进行操作。如需了解详情，请参阅 BigQuery Ruby API 参考文档。

在 GitHub 上查看反馈

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

返回页首

将数据流式插入到 BigQuery

准备工作

检查数据可用性

确保数据一致性

跨数据位置流式插入数据

流式插入到按提取时间分区的表

流式插入到分区表

使用模板表自动创建表

更改模板表架构

模板表详情

使用情况示例

大量事件日志

手动移除重复项

排查流式插入问题

流式插入示例

C#

Go

Java

Node.js

PHP

Python

Ruby

发送以下问题的反馈：