创建 Cloud Storage 订阅

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

准备工作

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

所需的角色和权限

以下是有关角色和权限的准则列表：

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

  • 如果您的订阅和主题位于不同的项目中，那么您还需要拥有资源级权限，如本部分稍后所述。

  • 如需创建 Cloud Storage 订阅，Pub/Sub 服务帐号必须具有向特定 Cloud Storage 存储桶写入数据以及读取存储桶元数据的权限。如需详细了解如何授予这些权限，请参阅本文档的下一部分。

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

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

所需权限

如需创建 Cloud Storage 订阅，您需要拥有以下权限：

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

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

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

为 Pub/Sub 服务帐号分配 Cloud Storage 角色

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

如需创建 Cloud Storage 订阅，Pub/Sub 服务帐号必须有权向特定 Cloud Storage 存储桶写入数据以及读取存储桶元数据。选择以下过程之一：

  • 在存储桶级别授予权限。在特定 Cloud Storage 存储桶上，向 Pub/Sub 服务帐号授予 Storage Object Creator (roles/storage.objectCreator) 角色和 Storage Legacy Bucket Reader (roles/storage.legacyBucketReader) 角色。

  • 如果您必须在项目级层授予角色，则可以改为向包含 Cloud Storage 存储桶的项目授予 Storage Admin (roles/storage.admin) 角色。将此角色授予 Pub/Sub 服务帐号。

存储桶权限

如需在存储桶级别授予 Storage Object Creator (roles/storage.objectCreator) 和 Storage Legacy Bucket Reader (roles/storage.legacyBucketReader) 角色，请执行以下步骤：

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

    转到 Cloud Storage

  2. 点击要向其中写入消息的 Cloud Storage 存储桶。

    此时会打开存储分区详情页面。

  3. 在存储桶详情页面中，点击配置标签页。

  4. 在权限 > 按主账号查看标签页中，点击授予访问权限。

    系统随即会打开授予访问权限页面。

  5. 在添加主账号部分中，输入您的 Pub/Sub 服务帐号的名称。

    服务帐号的格式为 service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com。例如，对于 PROJECT_NUMBER=112233445566 的项目，服务帐号的格式为 service-112233445566@gcp-sa-pubsub.iam.gserviceaccount.com。

  6. 在分配角色 > 选择角色下拉列表中，输入 Creator 并选择 Storage Object Creator 角色。

  7. 点击添加其他角色。

  8. 在选择角色下拉列表中，输入 Bucket Reader，然后选择 Storage Legacy Bucket Reader 角色。

  9. 点击保存。

项目权限

如需在项目级层授予 Storage Admin (roles/storage.admin) 角色，请执行以下步骤：

  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. 在分配角色 > 选择角色下拉列表中，输入 Storage Admin 并选择 Storage Admin 角色。

  5. 点击保存。

如需详细了解 Cloud Storage IAM，请参阅 Cloud Storage Identity and Access Management。

Cloud Storage 订阅属性

配置 Cloud Storage 订阅时，您必须指定所有订阅类型的通用属性以及一些额外的 Cloud Storage 订阅专用属性。

通用订阅属性

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

存储桶名称

在创建 Cloud Storage 订阅之前，Cloud Storage 存储桶必须已存在。

这些消息批量发送并存储在 Cloud Storage 存储桶中。 单个批次或文件作为对象存储在存储桶中。

Cloud Storage 存储桶必须停用请求者付款功能。

如需创建 Cloud Storage 存储桶，请参阅创建存储分区。

文件名前缀、后缀和日期时间

Cloud Storage 订阅生成的 Cloud Storage 输出文件作为对象存储在 Cloud Storage 存储桶中。存储在 Cloud Storage 存储桶中的对象的名称采用以下格式：<file-prefix><UTC-date-time>_<uuid><file-suffix>。

以下列表包含文件格式的详细信息以及您可以自定义的字段：

  • <file-prefix> 是自定义文件名前缀。这是一个可选字段。

  • <UTC-date-time> 是基于对象的创建时间自动生成的可自定义字符串。

  • <uuid> 是为对象自动生成的随机字符串。

  • <file-suffix> 是自定义文件名后缀。这是一个可选字段。文件名后缀不能以“/”结尾。

  • 您可以更改文件名前缀和后缀：

    • 例如，如果文件名前缀的值为 prod_，文件名后缀的值为 _archive，则示例对象名称为 prod_2023-09-25T04:10:00+00:00_uN1QuE_archive。

    • 如果您未指定文件名前缀和后缀，则存储在 Cloud Storage 存储桶中的对象名称将采用以下格式：<UTC-date-time>_<uuid>。

    • Cloud Storage 对象命名要求也适用于文件名前缀和后缀。如需了解详情，请参阅 Cloud Storage 对象简介。

  • 您可以更改文件名中日期和时间的显示方式：

    • 只能使用一次的必需日期时间匹配器：年（YYYY 或 YY）、月 (MM)、天 (DD)、小时 (hh)、分钟 (mm)、秒 (ss)。例如，YY-YYYY 或 MMM 无效。

    • 只能使用一次的可选匹配器：日期时间分隔符 (T) 和时区偏移量（Z 或 +00:00）。

    • 可以多次使用的可选元素：连字符 (-)、下划线 (_)、冒号 (:) 和正斜杠 (/)。

    • 例如，如果文件名日期时间格式的值为 YYYY-MM-DD/hh_mm_ssZ，则示例对象名称为 prod_2023-09-25/04_10_00Z_uNiQuE_archive。

    • 如果文件名日期时间格式以非匹配器字符结尾，则该字符将替换 <UTC-date-time> 和 <uuid> 之间的分隔符。例如，如果文件名日期时间格式的值为 YYYY-MM-DDThh_mm_ss-，则示例对象名称为 prod_2023-09-25T04_10_00-uNiQuE_archive。

文件批处理

借助 Cloud Storage 订阅，您可以决定何时要创建以对象形式存储在 Cloud Storage 存储桶中的新输出文件。当满足其中一个指定的批处理条件时，Pub/Sub 会写入输出文件。以下是 Cloud Storage 批处理条件：

  • 批量存储时长上限。这是一项必选设置。如果超出指定的最长时长值，则 Cloud Storage 订阅会写入新的输出文件。如果您未指定该值，系统会应用默认值 5 分钟。以下是适用于时长上限的值：

    • 最小值 = 1 分钟
    • 默认值 = 5 分钟
    • 最大值 = 10 分钟

  • 批量存储字节数上限。这是一项可选设置。如果超出指定的字节数上限值，Cloud Storage 订阅会写入新的输出文件。以下是适用的最大字节数值：

    • 最小值 = 1 KB
    • 最大值 = 10 GiB

例如，您可以将时长上限配置为 6 分钟，并将字节数上限配置为 2 GB。如果在第 4 分钟，输出文件的大小达到 2 GB，则 Pub/Sub 会最终确定上一个文件并开始写入新文件。

Cloud Storage 订阅可同时写入 Cloud Storage 存储桶中的多个文件。如果您已将订阅配置为每 6 分钟创建一个新文件，则可能会发现每 6 分钟会创建多个 Cloud Storage 文件。

在某些情况下，Pub/Sub 可能比文件批处理条件配置的时间更早开始写入新文件。如果订阅接收的消息大于最大字节数值，则文件可能也会超过最大字节数值。

文件格式

创建 Cloud Storage 订阅时，您可以将要存储在 Cloud Storage 存储桶中的输出文件指定为 Text 或 Avro 格式。

  • 文本：消息以纯文本形式存储。换行符将消息与文件中的上一条消息分隔开来。系统仅存储消息载荷，而不存储属性或其他元数据。

  • Avro：消息以 Apache Avro 二进制文件格式存储。

    如果选择 Avro，您还可以启用写入元数据选项。通过此选项，您可以存储邮件的元数据。

    subscription_name、message_id、publish_time 和属性字段等元数据会写入输出 Avro 对象中的顶级字段，而数据以外的所有其他消息属性（例如 order_key，如果存在）将作为条目添加到属性映射中。

    如果停用写入元数据，则只有消息载荷会写入输出 Avro 对象。

    以下是未启用写入元数据的输出消息的 Avro 架构：

    {
  "type": "record",
  "namespace": "com.google.pubsub",
  "name": "PubsubMessage",
  "fields": [
    { "name": "data", "type": "bytes" }
  ]
}

    以下是启用了写入元数据的输出消息的 Avro 架构：

    {
  "type": "record",
  "namespace": "com.google.pubsub",
  "name": "PubsubMessageWithMetadata",
  "fields": [
    { "name": "subscription_name", "type": "string" },
    { "name": "message_id", "type": "string"  },
    { "name": "publish_time", "type": {
        "type": "long",
        "logicalType": "timestamp-micros"
      }
    },
    { "name": "attributes", "type": { "type": "map", "values": "string" } },
    { "name": "data", "type": "bytes" }
  ]
}

创建 Cloud Storage 订阅

控制台

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

    前往“订阅”页面

  2. 点击创建订阅。

  3. 在订阅 ID 字段中，输入一个名称。

    如需了解如何为订阅命名，请参阅主题或订阅命名准则。

  4. 从下拉菜单中选择或创建一个主题。

    订阅将接收来自该主题的消息。

    如需了解如何创建主题，请参阅创建和管理主题。

  5. 在传送类型中选择写入 Cloud Storage。

  6. 对于 Cloud Storage 存储桶，点击浏览。

    • 您可以从任意适当的项目中选择现有存储桶。

    • 您还可以点击创建图标，然后按照屏幕上的说明创建新的存储桶。

      创建存储桶后，为 Cloud Storage 订阅选择存储桶。

      如需详细了解如何创建存储桶，请参阅创建存储桶。

    指定存储桶时，Pub/Sub 会检查存储桶中针对 Pub/Sub 服务帐号的相应权限。如果存在权限问题，您会看到类似于以下内容的消息：Unable to verify if the Pub/Sub service agent has write permissions on this bucket. You may be lacking permissions to view or set permissions。

  7. 如果遇到权限问题，请点击设置权限，然后按照屏幕上的说明操作。

    或者，请按照为 Pub/Sub 服务帐号分配 Cloud Storage 角色中的说明操作。

  8. 在文件格式部分，选择文本或 Avro。

    如果您选择 Avro，还可以视需要指定是否要在输出中存储消息元数据。

    如需详细了解这两个选项（包括 Avro 格式的消息元数据选项），请参阅文件格式。

  9. 可选：您可以为要写入 Cloud Storage 存储桶的所有文件指定文件名前缀和后缀。文件以对象的形式存储在存储桶中。

    如需详细了解如何设置文件前缀和文件后缀，请参阅文件前缀和后缀。

  10. 对于文件批处理，请指定创建新文件之前经过的最长时间。

    您也可以选择设置文件的大小上限。

    如需详细了解这两个文件批处理选项，请参阅文件批处理。

  11. 我们强烈建议您启用死信来处理消息失败情况。

    如需了解详情，请参阅死信主题。

  12. 您可以将其他设置保留为默认值，然后点击创建。

gcloud

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

    激活 Cloud Shell

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

  2. 如需创建 Cloud Storage 订阅，请运行 gcloud pubsub subscriptions create 命令。 
    gcloud pubsub subscriptions create SUBSCRIPTION_ID \
    --topic=TOPIC_ID \
    --cloud-storage-bucket=BUCKET_NAME \
    --cloud-storage-file-prefix=CLOUD_STORAGE_FILE_PREFIX \
    --cloud-storage-file-suffix=CLOUD_STORAGE_FILE_SUFFIX \
    --cloud-storage-max-bytes=CLOUD_STORAGE_MAX_BYTES \
    --cloud-storage-max-duration=CLOUD_STORAGE_MAX_DURATION \
    --cloud-storage-output-format=CLOUD_STORAGE_OUTPUT_FORMAT \
    --cloud-storage-write-metadata

    在命令中，只有 SUBSCRIPTION_ID、--topic 标志和 --cloud-storage-bucket 标志是必需的。其余标志是可选的，可以省略。

    请替换以下内容：

    • SUBSCRIPTION_ID：新 Cloud Storage 订阅的名称或 ID。
    • TOPIC_ID：主题的名称或 ID。
    • BUCKET_NAME：指定现有存储桶的名称。例如 prod_bucket。存储桶名称不得包含项目 ID。如需创建存储桶，请参阅创建存储桶。
    • CLOUD_STORAGE_FILE_PREFIX：指定 Cloud Storage 文件名的前缀。例如 log_events_。
    • CLOUD_STORAGE_FILE_SUFFIX：指定 Cloud Storage 文件名的后缀。例如 .txt。
    • CLOUD_STORAGE_MAX_BYTES：创建新文件之前可写入 Cloud Storage 文件的最大字节数。该值必须介于 1KB 到 10GB 之间。例如 20MB。
    • CLOUD_STORAGE_MAX_DURATION：在创建新的 Cloud Storage 文件之前可以经过的最长时长。该值必须介于 1m 到 10m 之间。例如 5m。
    • CLOUD_STORAGE_OUTPUT_FORMAT：写入 Cloud Storage 的数据的输出格式。具体值如下：
      • text：消息以原始文本形式编写，并用换行符分隔。
      • avro：消息作为 Avro 二进制文件编写。--cloud-storage-write-metadata 仅对输出格式为 avro 的订阅有效。

C++

在试用此示例之前，请按照 Pub/Sub 快速入门：使用客户端库中的 C++ 设置说明进行操作。如需了解详情，请参阅 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& bucket) {
  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_cloud_storage_config()->set_bucket(bucket);
  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";
}

Go

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

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

import (
	"context"
	"fmt"
	"io"
	"time"

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

// createCloudStorageSubscription creates a Pub/Sub subscription that exports messages to Cloud Storage.
func createCloudStorageSubscription(w io.Writer, projectID, subID string, topic *pubsub.Topic, bucket string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// topic of type https://godoc.org/cloud.google.com/go/pubsub#Topic
	// note bucket should not have the gs:// prefix
	// bucket := "my-bucket"
	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,
		CloudStorageConfig: pubsub.CloudStorageConfig{
			Bucket:         bucket,
			FilenamePrefix: "log_events_",
			FilenameSuffix: ".avro",
			OutputFormat:   &pubsub.CloudStorageOutputFormatAvroConfig{WriteMetadata: true},
			MaxDuration:    1 * time.Minute,
			MaxBytes:       1e8,
		},
	})
	if err != nil {
		return fmt.Errorf("client.CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created Cloud Storage subscription: %v\n", sub)

	return nil
}

Java

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

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

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.protobuf.Duration;
import com.google.pubsub.v1.CloudStorageConfig;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateCloudStorageSubscriptionExample {
  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 bucket = "your-bucket";
    String filenamePrefix = "log_events_";
    String filenameSuffix = ".text";
    Duration maxDuration = Duration.newBuilder().setSeconds(300).build();

    createCloudStorageSubscription(
        projectId, topicId, subscriptionId, bucket, filenamePrefix, filenameSuffix, maxDuration);
  }

  public static void createCloudStorageSubscription(
      String projectId,
      String topicId,
      String subscriptionId,
      String bucket,
      String filenamePrefix,
      String filenameSuffix,
      Duration maxDuration)
      throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

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

      CloudStorageConfig cloudStorageConfig =
          CloudStorageConfig.newBuilder()
              .setBucket(bucket)
              .setFilenamePrefix(filenamePrefix)
              .setFilenameSuffix(filenameSuffix)
              .setMaxDuration(maxDuration)
              .build();

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

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

监控订阅

Cloud Monitoring 提供了多种指标来监控订阅。

您还可以从 Pub/Sub 中监控订阅。

后续步骤