更改主题类型

您可以将导入主题转换为标准主题，也可以将标准主题转换为导入主题。

将导入主题转换为标准主题

如需将导入主题转换为标准主题，请清除提取设置。执行以下步骤：

控制台

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

    打开“主题”

  2. 点击导入主题。

  3. 在主题详情页面中，点击修改 。

  4. 清除启用提取 选项。

  5. 点击更新 。

gcloud

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

    激活 Cloud Shell

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

  2. 运行 gcloud pubsub topics update 命令：

    gcloud pubsub topics update TOPIC_ID \
    --clear-ingestion-data-source-settings

    将 TOPIC_ID 替换为主题 ID。

将标准主题转换为 Amazon Kinesis Data Streams 导入主题

如需将标准主题转换为 Amazon Kinesis Data Streams 导入主题， 请先检查您是否满足所有 的 前提条件。

控制台

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

    打开“主题”

  2. 点击要转换为导入主题的主题。

  3. 在主题详情页面中，点击修改 。

  4. 选择启用提取 选项。

  5. 对于提取来源，选择 Amazon Kinesis Data Streams 。

  6. 输入以下详细信息：

    • Kinesis 数据流 ARN：您计划注入到 Pub/Sub 中的 Kinesis Data Stream 的 ARN。ARN 格式如下：arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}。

    • Kinesis 使用方 ARN：已注册到 AWS Kinesis Data Stream 的使用方资源的 ARN。ARN 格式如下：arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}。

    • AWS 角色 ARN：AWS 角色的 ARN。角色的 ARN 格式如下： arn:aws:iam::${Account}:role/${RoleName}.

    • 服务账号：您创建的服务账号。

  7. 点击更新 。

gcloud

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

    激活 Cloud Shell

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

  2. 运行 gcloud pubsub topics update 命令，并使用以下 示例中提及的所有标志：

    gcloud pubsub topics update TOPIC_ID 
    
  --kinesis-ingestion-stream-arn KINESIS_STREAM_ARN
    
  --kinesis-ingestion-consumer-arn KINESIS_CONSUMER_ARN
    
  --kinesis-ingestion-role-arn KINESIS_ROLE_ARN
    
  --kinesis-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

    替换以下内容：

    • TOPIC_ID 是主题 ID 或名称。此字段无法更新。

    • KINESIS_STREAM_ARN 是您计划注入到 Pub/Sub 中的 Kinesis Data Streams 的 ARN。ARN 格式如下： arn:${Partition}:kinesis:${Region}:${Account}:stream/${StreamName}。

    • KINESIS_CONSUMER_ARN 是已注册到 AWS Kinesis Data Streams 的使用方资源的 ARN 。ARN 格式如下：arn:${Partition}:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimestamp}。

    • KINESIS_ROLE_ARN 是 AWS 角色的 ARN。角色的 ARN 格式如下： arn:aws:iam::${Account}:role/${RoleName}。

    • PUBSUB_SERVICE_ACCOUNT 是您 创建的服务帐号。

Go

以下示例使用 Go Pub/Sub 客户端库的主要版本 (v2)。如果您仍在使用 v1 库，请参阅 迁移到 v2 的指南。 如需查看 v1 代码示例列表，请参阅 已废弃的代码示例。

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

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
	"google.golang.org/protobuf/types/known/fieldmaskpb"
)

func updateTopicType(w io.Writer, projectID, topic string) error {
	// projectID := "my-project-id"
	// topic := "projects/my-project-id/topics/my-topic"
	streamARN := "stream-arn"
	consumerARN := "consumer-arn"
	awsRoleARN := "aws-role-arn"
	gcpServiceAccount := "gcp-service-account"

	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	pbTopic := &pubsubpb.Topic{
		Name: topic,
		IngestionDataSourceSettings: &pubsubpb.IngestionDataSourceSettings{
			Source: &pubsubpb.IngestionDataSourceSettings_AwsKinesis_{
				AwsKinesis: &pubsubpb.IngestionDataSourceSettings_AwsKinesis{
					StreamArn:         streamARN,
					ConsumerArn:       consumerARN,
					AwsRoleArn:        awsRoleARN,
					GcpServiceAccount: gcpServiceAccount,
				},
			},
		},
	}
	updateReq := &pubsubpb.UpdateTopicRequest{
		Topic: pbTopic,
		UpdateMask: &fieldmaskpb.FieldMask{
			Paths: []string{"ingestion_data_source_settings"},
		},
	}
	topicCfg, err := client.TopicAdminClient.UpdateTopic(ctx, updateReq)
	if err != nil {
		return fmt.Errorf("topic.Update: %w", err)
	}
	fmt.Fprintf(w, "Topic updated with kinesis source: %v\n", topicCfg)
	return nil
}

Java

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


import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.protobuf.FieldMask;
import com.google.pubsub.v1.IngestionDataSourceSettings;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import com.google.pubsub.v1.UpdateTopicRequest;
import java.io.IOException;

public class UpdateTopicTypeExample {
  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";
    // Kinesis ingestion settings.
    String streamArn = "stream-arn";
    String consumerArn = "consumer-arn";
    String awsRoleArn = "aws-role-arn";
    String gcpServiceAccount = "gcp-service-account";

    UpdateTopicTypeExample.updateTopicTypeExample(
        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
  }

  public static void updateTopicTypeExample(
      String projectId,
      String topicId,
      String streamArn,
      String consumerArn,
      String awsRoleArn,
      String gcpServiceAccount)
      throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      TopicName topicName = TopicName.of(projectId, topicId);

      IngestionDataSourceSettings.AwsKinesis awsKinesis =
          IngestionDataSourceSettings.AwsKinesis.newBuilder()
              .setStreamArn(streamArn)
              .setConsumerArn(consumerArn)
              .setAwsRoleArn(awsRoleArn)
              .setGcpServiceAccount(gcpServiceAccount)
              .build();
      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();

      // Construct the topic with Kinesis ingestion settings.
      Topic topic =
          Topic.newBuilder()
              .setName(topicName.toString())
              .setIngestionDataSourceSettings(ingestionDataSourceSettings)
              .build();

      // Construct a field mask to indicate which field to update in the topic.
      FieldMask updateMask =
          FieldMask.newBuilder().addPaths("ingestion_data_source_settings").build();

      UpdateTopicRequest request =
          UpdateTopicRequest.newBuilder().setTopic(topic).setUpdateMask(updateMask).build();

      Topic response = topicAdminClient.updateTopic(request);

      System.out.println(
          "Updated topic with Kinesis ingestion settings: " + response.getAllFields());
    }
  }
}

Node.js

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

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const awsRoleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

// 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 updateTopicIngestionType(
  topicNameOrId,
  awsRoleArn,
  gcpServiceAccount,
  streamArn,
  consumerArn,
) {
  const metadata = {
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  };

  await pubSubClient.topic(topicNameOrId).setMetadata(metadata);

  console.log('Topic updated with Kinesis source successfully.');
}

Python

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

from google.cloud import pubsub_v1
from google.pubsub_v1.types import Topic
from google.pubsub_v1.types import IngestionDataSourceSettings
from google.pubsub_v1.types import UpdateTopicRequest
from google.protobuf import field_mask_pb2

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# stream_arn = "your-stream-arn"
# consumer_arn = "your-consumer-arn"
# aws_role_arn = "your-aws-role-arn"
# gcp_service_account = "your-gcp-service-account"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

update_request = UpdateTopicRequest(
    topic=Topic(
        name=topic_path,
        ingestion_data_source_settings=IngestionDataSourceSettings(
            aws_kinesis=IngestionDataSourceSettings.AwsKinesis(
                stream_arn=stream_arn,
                consumer_arn=consumer_arn,
                aws_role_arn=aws_role_arn,
                gcp_service_account=gcp_service_account,
            )
        ),
    ),
    update_mask=field_mask_pb2.FieldMask(paths=["ingestion_data_source_settings"]),
)

topic = publisher.update_topic(request=update_request)
print(f"Updated topic: {topic.name} with AWS Kinesis Ingestion Settings")

C++

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

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::TopicAdminClient client, std::string project_id,
   std::string topic_id, std::string stream_arn, std::string consumer_arn,
   std::string aws_role_arn, std::string gcp_service_account) {
  google::pubsub::v1::UpdateTopicRequest request;

  request.mutable_topic()->set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto* aws_kinesis = request.mutable_topic()
                          ->mutable_ingestion_data_source_settings()
                          ->mutable_aws_kinesis();
  aws_kinesis->set_stream_arn(stream_arn);
  aws_kinesis->set_consumer_arn(consumer_arn);
  aws_kinesis->set_aws_role_arn(aws_role_arn);
  aws_kinesis->set_gcp_service_account(gcp_service_account);
  *request.mutable_update_mask()->add_paths() =
      "ingestion_data_source_settings";

  auto topic = client.UpdateTopic(request);
  if (!topic) throw std::move(topic).status();

  std::cout << "The topic was successfully updated: " << topic->DebugString()
            << "\n";
}

Node.js

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

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const awsRoleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';
// const streamArn = 'arn:aws:kinesis:...';
// const consumerArn = 'arn:aws:kinesis:...';

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

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

async function updateTopicIngestionType(
  topicNameOrId: string,
  awsRoleArn: string,
  gcpServiceAccount: string,
  streamArn: string,
  consumerArn: string,
) {
  const metadata: TopicMetadata = {
    ingestionDataSourceSettings: {
      awsKinesis: {
        awsRoleArn,
        gcpServiceAccount,
        streamArn,
        consumerArn,
      },
    },
  };

  await pubSubClient.topic(topicNameOrId).setMetadata(metadata);

  console.log('Topic updated with Kinesis source successfully.');
}

如需详细了解 ARN，请参阅 Amazon 资源名称 (ARN) 和 IAM 标识符。

将标准主题转换为 Cloud Storage 导入主题

如需将标准主题转换为 Cloud Storage 导入主题， 请先检查您是否满足所有 前提条件。

控制台

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

    打开“主题”

  2. 点击要转换为 Cloud Storage 导入主题的主题。

  3. 在主题详情页面中，点击修改 。

  4. 选择启用提取 选项。

  5. 对于提取来源，选择 Google Cloud Storage 。

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

    此时会打开选择存储桶 页面。从下列选项中选择一项：

    • 从任何合适的项目中选择现有存储桶。

    • 点击创建图标，然后按照屏幕上的说明 创建新存储桶。创建存储桶后，请为 Cloud Storage 导入主题选择存储桶 。

  7. 指定存储桶后，Pub/Sub 会检查 Pub/Sub 服务帐号对该存储桶是否具有 适当的权限。如果存在权限问题，您会看到与权限相关的错误消息 。

    如果遇到权限问题，请点击设置权限 。如需了解详情，请参阅 向 Cloud Storage 服务账号授予 Pub/Sub 权限。

  8. 对于对象格式，请选择文本、Avro或Pub/Sub Avro。

    如果您选择文本 ，可以选择指定一个分隔符 ，用于将对象拆分为消息。

    如需详细了解这些选项，请参阅输入格式。

  9. 可选。您可以为您的 主题指定创建对象的最短时间。如果设置了此选项，系统只会提取在创建对象的最短时间之后创建的对象 。

    如需了解详情，请参阅 创建对象的最短时间。

  10. 您必须指定 Glob 模式 。如需注入存储桶中的所有对象， 请使用 ** 作为 glob 模式。系统只会注入与给定模式 匹配的对象。

    如需了解详情， 请参阅 匹配 glob 模式。

  11. 保留其他默认设置。
  12. 点击更新主题 。

gcloud

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

    激活 Cloud Shell

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

  2. 为避免丢失导入主题的设置，请确保 每次更新主题时都包含所有设置。如果您遗漏了某些设置 ，Pub/Sub 会将这些设置重置为其原始默认值。

    运行 gcloud pubsub topics update 命令，并使用以下 示例中提及的所有标志：

    gcloud pubsub topics update TOPIC_ID \
    --cloud-storage-ingestion-bucket=BUCKET_NAME\
    --cloud-storage-ingestion-input-format=INPUT_FORMAT\
    --cloud-storage-ingestion-text-delimiter=TEXT_DELIMITER\
    --cloud-storage-ingestion-minimum-object-create-time=MINIMUM_OBJECT_CREATE_TIME\
    --cloud-storage-ingestion-match-glob=MATCH_GLOB

    替换以下内容：

    • TOPIC_ID 是主题 ID 或名称。此字段无法更新。

    • BUCKET_NAME：指定现有存储桶的名称。 例如 prod_bucket。 存储桶名称不得包含项目 ID。 如需创建存储桶，请参阅创建存储桶。

    • INPUT_FORMAT：指定要注入的对象格式。 此值可以是 text、avro 或 pubsub_avro。 如需详细了解这些选项，请参阅 输入格式。

    • TEXT_DELIMITER：指定用于将 文本对象拆分为 Pub/Sub 消息的分隔符。 此值必须为单个字符，并且仅当 INPUT_FORMAT 为 text 时才必须设置。 默认值为换行符 (\n)。

      使用 gcloud CLI 指定分隔符时，请密切注意换行符 \n 等特殊字符的处理方式。请使用 '\n' 格式，以确保定界符得到正确解读。如果仅使用 \n 而不加引号或转义，则分隔符将为 "n"。

    • MINIMUM_OBJECT_CREATE_TIME：指定对象必须在何时创建才能被注入。 此值应采用 UTC 时间，格式为 YYYY-MM-DDThh:mm:ssZ。 例如 2024-10-14T08:30:30Z。

      从 0001-01-01T00:00:00Z 到 9999-12-31T23:59:59Z（含）之间的任何日期（过去或将来）均有效。

    • MATCH_GLOB：指定要匹配的 glob 模式，以便 注入对象。使用 gcloud CLI 时， 包含 * 字符的匹配 glob 必须将 * 字符格式化为转义字符，形式为 \*\*.txt ，或者整个匹配 glob 必须用引号括起来，形式为 "**.txt" 或 '**.txt'。 如需了解 glob 模式支持的语法，请参阅 Cloud Storage 文档。