创建 BigQuery 订阅

本文档介绍了如何创建 BigQuery 订阅。 您可以使用 Google Cloud 控制台、Google Cloud CLI、客户端库 或使用 Pub/Sub API 创建 BigQuery 订阅。

准备工作

在阅读本文档之前，请确保您熟悉以下内容：

除了熟悉 Pub/Sub BigQuery，请确保您满足以下前提条件 在创建 BigQuery 订阅之前：

  • BigQuery 表已存在。或者，您可以创建一个 请按照下方说明创建 BigQuery 订阅， 请参阅本文档后面的部分。

  • Pub/Sub 主题的架构与 BigQuery 表。如果您添加不兼容的 您收到与兼容性相关的错误 消息。如需了解详情，请参阅 架构兼容性。

所需的角色和权限

以下是有关角色和权限的一系列指南：

  • 如需创建订阅，您必须在项目中配置访问权限控制 。

  • 如果您的 订阅和主题属于不同的项目，稍后讨论 部分。

  • 要创建 BigQuery 订阅， Pub/Sub 服务账号必须具有写入 特定的 BigQuery 表如需详细了解如何 具体请参阅本文档的下一部分。

  • 您可以在项目中配置 BigQuery 订阅 才能写入其他项目中的 BigQuery 表。

如需获取创建 BigQuery 订阅所需的权限， 请让管理员向您授予 项目的 Pub/Sub Editor (roles/pubsub.editor) IAM 角色。 如需详细了解如何授予角色，请参阅管理访问权限。

此预定义角色包含 创建 BigQuery 订阅所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

创建 BigQuery 订阅需要以下权限：

  • 从订阅中拉取： pubsub.subscriptions.consume
  • 创建订阅： pubsub.subscriptions.create
  • 删除订阅： pubsub.subscriptions.delete
  • 获取订阅： pubsub.subscriptions.get
  • 列出订阅： pubsub.subscriptions.list
  • 更新订阅： pubsub.subscriptions.update
  • 将订阅附加到主题： pubsub.topics.attachSubscription
  • 获取订阅的 IAM 政策： pubsub.subscriptions.getIamPolicy
  • 为订阅配置 IAM 政策： pubsub.subscriptions.setIamPolicy

您也可以使用自定义角色或其他预定义角色来获取这些权限。

如果您需要创建 BigQuery 一个项目中的订阅与另一个项目中的主题关联 项目，请让主题管理员也授予您 Pub/Sub Editor (roles/pubsub.editor) IAM 角色。

为 Pub/Sub 服务账号分配 BigQuery 角色

某些 Google Cloud 服务具有由 Google Cloud 代管式服务账号， 服务访问您的资源。这些服务账号 称为服务代理Pub/Sub 会创建并维护 为每个项目创建服务账号，格式为 service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com.

为了创建 BigQuery 订阅，Pub/Sub 服务账号必须拥有写入权限 并读取表元数据。

授予 BigQuery 数据编辑者 (roles/bigquery.dataEditor) 授予 Pub/Sub 服务账号的权限。

  1. 在 Google Cloud 控制台中，转到 IAM 页面。

    转到 IAM

  2. 点击授予访问权限。

  3. 在添加主账号部分中，输入 Pub/Sub 的名称。 服务账号。服务账号的格式为 service-project-number@gcp-sa-pubsub.iam.gserviceaccount.com。例如，对于具有 project-number=112233445566 的项目，服务账号的格式为 service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com。

  4. 在分配角色部分中，点击添加其他角色。

  5. 在选择角色下拉菜单中，输入 BigQuery。 然后选择 BigQuery Data Editor 角色。

  6. 点击保存。

如需详细了解 BigQuery IAM 请参阅 BigQuery 角色和权限。

BigQuery 订阅属性

配置 BigQuery 订阅时，您可以指定以下内容 属性。

通用属性

了解常见的订阅属性 您可以为所有订阅设置此折扣

使用主题架构

此选项允许 Pub/Sub 使用架构 Pub/Sub 主题 已附加订阅。此外，Pub/Sub 将消息中的字段写入 BigQuery 表中的列。

使用此选项时，请务必检查以下额外要求：

  • 主题架构和 BigQuery 架构中的字段 必须具有相同的名称，并且其类型必须相互兼容。

  • 主题架构中的任何选填字段也必须为 在 BigQuery 架构中是可选属性。

  • 主题架构中的必填字段不必是 BigQuery 架构中必需的参数。

  • 如果存在 BigQuery 字段不存在， 主题架构，这些 BigQuery 字段 必须采用 NULLABLE 模式。

  • 如果主题架构包含 不存在于 BigQuery 架构中，而 字段，请选择丢弃未知字段选项。

  • 您只能选择其中一个订阅属性：使用主题架构 或使用表架构。

如果您不选择使用主题架构或使用表架构选项， 确保 BigQuery 表有一个名为 data 的列 类型 BYTES、STRING 或 JSON。Pub/Sub 会将消息写入到 此 BigQuery 列。

您可能看不到 Pub/Sub 主题架构更改或 BigQuery 表架构会在消息中立即生效 写入 BigQuery 表。例如，如果拖放 未知字段选项，且 Pub/Sub 架构（而非 BigQuery 架构） 写入 BigQuery 表的消息可能仍然不包含 该字段添加到 BigQuery 架构中。最终， 架构同步，后续消息包含 字段。

对 BigQuery 使用使用主题架构选项时 还可以利用 BigQuery 变更 数据捕获 (CDC)。CDC 通过以下方式更新 BigQuery 表： 处理更改并将其应用到现有行。

如需详细了解此功能，请参阅使用变更数据捕获流式传输表更新。

如需了解如何将此功能与 BigQuery 订阅搭配使用， 请参阅 BigQuery 变更数据捕获。

使用表架构

通过此选项，Pub/Sub 可以使用 用于写入 JSON 字段的 BigQuery 表 消息发送到相应的列。使用此选项时，请注意 请查看以下额外要求：

  • 已发布的消息必须采用 JSON 格式。

  • 如果订阅的主题 具有关联的架构 消息编码属性必须设置为 JSON。

  • 如果存在 BigQuery 字段不存在， 消息，这些 BigQuery 字段必须采用 NULLABLE 模式。

  • 如果邮件中包含 可以删除 BigQuery 架构和这些字段，请选择 选项丢弃未知字段。

  • 在 JSON 消息中，DATE、DATETIME、TIME 和 TIMESTAMP 值 必须是符合受支持表示法的整数。

  • 在 JSON 消息中，NUMERIC 和 BIGNUMERIC 值必须使用 BigDecimalByteStringEncoder。

  • 您只能选择其中一个订阅属性：使用主题架构 或使用表架构。

如果您不选择使用主题架构或使用表架构选项， 确保 BigQuery 表有一个名为 data 的列 类型 BYTES、STRING 或 JSON。Pub/Sub 会将消息写入到 此 BigQuery 列。

您可能不会看到 BigQuery 表架构的更改 将消息写入 BigQuery 表后立即生效。 例如，如果启用了删除未知字段选项，但有一个字段 存在于消息中，但不存在于 BigQuery 架构中， 写入 BigQuery 表的消息可能仍然不包含 该字段添加到 BigQuery 架构中。最终， 架构同步，后续消息中会包含该字段。

针对 BigQuery 订阅使用使用表架构选项时，您需要 还可以利用 BigQuery 变更数据捕获 (CDC)。 CDC 通过处理更改并将其应用于现有表来更新 BigQuery 表 行。

如需详细了解此功能，请参阅使用变更数据捕获流式传输表更新。

如需了解如何将此功能与 BigQuery 订阅搭配使用，请参阅 BigQuery 变更数据捕获。

删除未知字段

此选项与使用主题架构或使用表架构搭配使用 选项。此选项可让 Pub/Sub 删除主题中存在的任何字段 但不包含在 BigQuery 架构中。不含未知丢弃 字段设置，则包含额外字段的消息不会被写入到 BigQuery 并保留在订阅积压消息中。通过 订阅将处于错误状态。

写入元数据

此选项可让 Pub/Sub 将每条消息的元数据写入 BigQuery 表。否则，元数据 不会写入 BigQuery 表。

如果您选择写入元数据选项，请确保 BigQuery 表包含下表中所述的字段。

如果您不选择写入元数据选项，则目标 BigQuery 表仅需要 data 字段，除非 use_topic_schema 为 true。如果您同时选择了 Write metadata 和 使用主题架构选项，则主题的架构必须 不包含任何名称与元数据参数匹配的字段。 此限制包括这些蛇形参数的驼峰式大小写版本。

参数
subscription_name

STRING

订阅的名称。
message_id

STRING

消息的 ID
publish_time

TIMESTAMP

发布消息的时间。
data

BYTES、STRING 或 JSON

消息正文。

所有目标平台都必须填写 data 字段 未选择使用 主题架构或使用表架构。如果 字段为 JSON 类型，则消息正文必须为 有效的 JSON。
attributes

STRING 或 JSON

包含所有消息属性的 JSON 对象。它还 包含属于 包含排序键的 Pub/Sub 消息。 （如果存在）。

创建 BigQuery 订阅

以下示例演示了如何创建使用 BigQuery 传送功能的订阅。

控制台

  1. 在 Google Cloud 控制台中，前往 订阅页面。

    前往 订阅

  2. 点击创建订阅。
  3. 对于订阅 ID 字段，输入名称。

    对于 请参阅命名准则 主题或订阅。

  4. 从下拉菜单中选择或创建一个主题。订阅 从主题接收消息。
  5. 将传送类型选为写入 BigQuery。
  6. 为 BigQuery 表选择项目。
  7. 选择现有数据集或创建新数据集。

    相关信息 如需了解如何创建数据集，请参阅创建数据集。

  8. 选择现有表或创建一个新表。

    如需了解 如何创建表，请参阅创建表。

  9. 我们强烈建议您启用 Dead 字母以处理消息故障。

    有关详情，请参阅终止 字母主题。

  10. 点击创建。

您还可以从主题页面创建订阅。 此快捷方式可帮助您将主题与订阅关联。

  1. 在 Google Cloud 控制台中，前往主题 页面。

    前往 主题

  2. 点击您要查看的主题旁边的 。 以创建订阅。
  3. 从上下文菜单中，选择创建订阅。
  4. 将传送类型选为写入 BigQuery。
  5. 为 BigQuery 表选择项目。
  6. 选择现有数据集或创建新数据集。

    相关信息 如需了解如何创建数据集，请参阅创建数据集。

  7. 选择现有表或创建一个新表。

    如需了解 请参阅创建表。

  8. 我们强烈建议您启用 Dead 字母以处理消息故障。

    有关详情，请参阅终止 字母主题。

  9. 点击创建。

gcloud

  1. 在 Google Cloud 控制台中，激活 Cloud Shell。

    激活 Cloud Shell

    Cloud Shell 会话随即会在 Google Cloud 控制台的底部启动，并显示命令行提示符。Cloud Shell 是一个已安装 Google Cloud CLI 且已为当前项目设置值的 Shell 环境。该会话可能需要几秒钟时间来完成初始化。

  2. 如需创建 Pub/Sub 订阅，请使用 gcloud pubsub subscriptions create 命令：

    gcloud pubsub subscriptions create SUBSCRIPTION_ID \
    --topic=TOPIC_ID \
    --bigquery-table=PROJECT_ID:DATASET_ID.TABLE_ID

    替换以下内容：

    • SUBSCRIPTION_ID：指定 订阅。
    • TOPIC_ID：指定主题的 ID。通过 主题需要架构。
    • PROJECT_ID：指定项目的 ID。
    • DATASET_ID：指定现有 ID 数据集。要创建数据集，请参阅 创建 数据集。
    • TABLE_ID：指定现有表的 ID。 如果您的主题，则此表需要 data 字段 没有架构如需创建表，请参阅创建 具有架构定义的空表。

C++

在试用此示例之前，请按照C++ Pub/Sub 快速入门： 客户端库。 有关详情，请参阅 Pub/Sub C++ API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为客户端库设置身份验证。 

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id, std::string const& table_id) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.mutable_bigquery_config()->set_table(table_id);
  auto sub = client.CreateSubscription(request);
  if (!sub) {
    if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
      std::cout << "The subscription already exists\n";
      return;
    }
    throw std::move(sub).status();
  }

  std::cout << "The subscription was successfully created: "
            << sub->DebugString() << "\n";
}

C#

在试用此示例之前，请按照C# Pub/Sub 快速入门： 客户端库。 有关详情，请参阅 Pub/Sub C# API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为客户端库设置身份验证。 


using Google.Cloud.PubSub.V1;

public class CreateBigQuerySubscriptionSample
{
    public Subscription CreateBigQuerySubscription(string projectId, string topicId, string subscriptionId, string bigqueryTableId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            BigqueryConfig = new BigQueryConfig
            {
                Table = bigqueryTableId
            }
        };
        var subscription = subscriber.CreateSubscription(subscriptionRequest);
        return subscription;
    }
}

Go

在试用此示例之前，请按照Go Pub/Sub 快速入门： 客户端库。 有关详情，请参阅 Pub/Sub Go API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为客户端库设置身份验证。 

import (
	"context"
	"fmt"
	"io"

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

// createBigQuerySubscription creates a Pub/Sub subscription that exports messages to BigQuery.
func createBigQuerySubscription(w io.Writer, projectID, subID string, topic *pubsub.Topic, table string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// topic of type https://godoc.org/cloud.google.com/go/pubsub#Topic
	// table := "my-project-id.dataset_id.table_id"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	sub, err := client.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
		Topic: topic,
		BigQueryConfig: pubsub.BigQueryConfig{
			Table:         table,
			WriteMetadata: true,
		},
	})
	if err != nil {
		return fmt.Errorf("client.CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created BigQuery subscription: %v\n", sub)

	return nil
}

Java

在试用此示例之前，请按照Java Pub/Sub 快速入门： 客户端库。 有关详情，请参阅 Pub/Sub Java API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为客户端库设置身份验证。 

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.BigQueryConfig;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateBigQuerySubscriptionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";
    String bigqueryTableId = "your-project.your-dataset.your-table";

    createBigQuerySubscription(projectId, topicId, subscriptionId, bigqueryTableId);
  }

  public static void createBigQuerySubscription(
      String projectId, String topicId, String subscriptionId, String bigqueryTableId)
      throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      BigQueryConfig bigqueryConfig =
          BigQueryConfig.newBuilder().setTable(bigqueryTableId).setWriteMetadata(true).build();

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  .setBigqueryConfig(bigqueryConfig)
                  .build());

      System.out.println("Created a BigQuery subscription: " + subscription.getAllFields());
    }
  }
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const bigqueryTableId = 'YOUR_TABLE_ID';

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

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createBigQuerySubscription(
  topicNameOrId,
  subscriptionNameOrId,
  bigqueryTableId
) {
  const options = {
    bigqueryConfig: {
      table: bigqueryTableId,
      writeMetadata: true,
    },
  };

  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, options);

  console.log(`Subscription ${subscriptionNameOrId} created.`);
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const bigqueryTableId = 'YOUR_TABLE_ID';

// Imports the Google Cloud client library
import {PubSub, CreateSubscriptionOptions} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createBigQuerySubscription(
  topicNameOrId: string,
  subscriptionNameOrId: string,
  bigqueryTableId: string
) {
  const options: CreateSubscriptionOptions = {
    bigqueryConfig: {
      table: bigqueryTableId,
      writeMetadata: true,
    },
  };

  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, options);

  console.log(`Subscription ${subscriptionNameOrId} created.`);
}

PHP

在试用此示例之前，请按照PHP Pub/Sub 快速入门： 客户端库。 有关详情，请参阅 Pub/Sub PHP API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为客户端库设置身份验证。 

use Google\Cloud\PubSub\PubSubClient;
use Google\Cloud\PubSub\V1\BigQueryConfig;

/**
 * Creates a Pub/Sub BigQuery subscription.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 * @param string $table      The BigQuery table to which to write.
 */
function create_bigquery_subscription($projectId, $topicName, $subscriptionName, $table)
{
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);
    $topic = $pubsub->topic($topicName);
    $subscription = $topic->subscription($subscriptionName);
    $config = new BigQueryConfig(['table' => $table]);
    $subscription->create([
        'bigqueryConfig' => $config
    ]);

    printf('Subscription created: %s' . PHP_EOL, $subscription->name());
}

Python

在试用此示例之前，请按照Python Pub/Sub 快速入门： 客户端库。 有关详情，请参阅 Pub/Sub Python API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为客户端库设置身份验证。 

from google.cloud import pubsub_v1

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"
# bigquery_table_id = "your-project.your-dataset.your-table"

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

bigquery_config = pubsub_v1.types.BigQueryConfig(
    table=bigquery_table_id, write_metadata=True
)

# Wrap the subscriber in a 'with' block to automatically call close() to
# close the underlying gRPC channel when done.
with subscriber:
    subscription = subscriber.create_subscription(
        request={
            "name": subscription_path,
            "topic": topic_path,
            "bigquery_config": bigquery_config,
        }
    )

print(f"BigQuery subscription created: {subscription}.")
print(f"Table for subscription is: {bigquery_table_id}")

Ruby

在试用此示例之前，请按照Ruby Pub/Sub 快速入门： 客户端库。 有关详情，请参阅 Pub/Sub Ruby API 参考文档。

如需向 Pub/Sub 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为客户端库设置身份验证。 

require "google/cloud/pubsub"

##
# Shows how to create a BigQuery subscription where messages published
# to a topic populates a BigQuery table.
#
# @param project_id [String]
# Your Google Cloud project (e.g. "my-project")
# @param topic_id [String]
# Your topic name (e.g. "my-secret")
# @param subscription_id [String]
# ID for new subscription to be created (e.g. "my-subscription")
# @param bigquery_table_id [String]
# ID of bigquery table (e.g "my-project:dataset-id.table-id")
#
def pubsub_create_bigquery_subscription project_id:, topic_id:, subscription_id:, bigquery_table_id:
  pubsub = Google::Cloud::Pubsub.new project_id: project_id
  topic = pubsub.topic topic_id
  subscription = topic.subscribe subscription_id,
                                 bigquery_config: {
                                   table: bigquery_table_id,
                                   write_metadata: true
                                 }
  puts "BigQuery subscription created: #{subscription_id}."
  puts "Table for subscription is: #{bigquery_table_id}"
end

后续步骤