使用 SMT 创建订阅

本文档介绍如何使用单条消息转换 (SMT) 创建 Pub/Sub 订阅。

借助订阅 SMT，可以直接在 Pub/Sub 中对消息数据和属性进行轻量级修改。此功能可在消息传送给订阅者客户端之前，对数据进行清理、过滤或格式转换。

如需创建具有 SMT 的订阅，您可以使用 Google Cloud 控制台、Google Cloud CLI、客户端库或 Pub/Sub API。

准备工作

所需的角色和权限

如需获得创建具有 SMT 的订阅所需的权限，请让您的管理员为您授予项目的 Pub/Sub Editor (roles/pubsub.editor) IAM 角色。 如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

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

所需权限

如需创建包含 SMT 的订阅，您需要具备以下权限：

  • 授予在项目上创建订阅的权限： pubsub.subscriptions.create

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

根据订阅类型，您可能需要其他权限。 如需了解确切的权限列表，请参阅讨论如何创建特定订阅的文档。例如，如果您要创建具有 SMT 的 BigQuery 订阅，请参阅创建 BigQuery 订阅页面。

如果您在与主题不同的项目中创建订阅，则必须在包含主题的项目中向包含订阅的项目的主账号授予 roles/pubsub.subscriber 角色。

您可以在项目级层和个别资源级层配置访问权限控制。

创建包含 SMT 的订阅

在创建包含 SMT 的订阅之前，请先查看订阅的属性文档。

如需创建包含一个或多个 SMT 的 Pub/Sub 订阅，请执行以下步骤。

控制台

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

    前往订阅页面

  2. 点击创建订阅。

    系统会打开创建订阅页面。

  3. 在订阅 ID 字段中，输入订阅的 ID。如需详细了解如何命名订阅，请参阅命名准则。

  4. 在转换下，点击添加转换。

  5. 输入函数名称。例如：redactSSN。

  6. 如果您不希望 SMT 立即处于有效状态，请选择停用转换。选择此选项后，系统会使用订阅创建 SMT，但不会针对收到的邮件执行该 SMT。创建订阅后，您可以修改订阅以启用 SMT。

  7. 在文本区域中，输入 SMT 的代码。例如：

    function redactSSN(message, metadata) {
  const data = JSON.parse(message.data);
  delete data['ssn'];
  message.data = JSON.stringify(data);
  return message;
}

  8. 可选。如需验证 SMT，请点击验证。如果 SMT 有效，系统会显示消息 "Validation passed"。否则，系统会显示错误消息。

  9. 如需添加其他转换，请点击添加转换，然后重复上述步骤。

    如需按特定顺序排列 SMT，请点击上移或下移。如需移除 SMT，请点击 Delete（删除）。

  10. 可选。如需在示例消息上测试 SMT，请执行以下步骤：

    1. 点击 Test transforms。

    2. 在测试转换窗口中，选择要测试的函数。

    3. 在输入消息窗口中，输入示例消息。

    4. 如需向消息添加属性，请点击添加属性，然后输入属性的键和值。您可以添加多个属性。

    5. 点击测试。应用 SMT 后的消息结果会显示在输出消息下。

    6. 如需关闭测试转换窗口，请点击 关闭。

    如果您创建了多个 SMT，可以按如下方式测试整个转换序列：

    1. 按照上述步骤中的说明，测试序列中的第一个 SMT。
    2. 选择下一个 SMT。输入消息已预先填充了上一次测试的输出消息。
    3. 继续按顺序测试 SMT，以确保整个序列按预期运行。

  11. 点击创建以创建订阅。

gcloud

  1. In the Google Cloud console, activate Cloud Shell.

    Activate Cloud Shell

    At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

  2. 创建用于定义一个或多个 SMT 的 YAML 或 JSON 文件。如果您有多个 SMT，系统会按您列出的顺序对消息执行这些 SMT。

    以下是 YAML 转换文件的示例：

    - javascriptUdf:
    code: >
        function redactSSN(message, metadata) {
          const data = JSON.parse(message.data);
          delete data['ssn'];
          message.data = JSON.stringify(data);
          return message;
        }
    functionName: redactSSN

  3. 可选。如需验证 SMT，请运行 gcloud pubsub message-transforms validate 命令：

    gcloud pubsub message-transforms validate \
  --message-transform-file=TRANSFORM_FILE

    替换以下内容：

    • TRANSFORM_FILE：定义单个 SMT 的 YAML 或 JSON 文件的路径。如果您要创建多个 SMT，则必须分别验证它们。

  4. 可选。如需针对示例 Pub/Sub 消息测试一个或多个 SMT，请运行 gcloud pubsub message-transforms test 命令：

    gcloud pubsub message-transforms test \
  --message-transforms-file=TRANSFORMS_FILE \
  --message=MESSAGE \
  --attribute=ATTRIBUTES

    替换以下内容：

    • TRANSFORMS_FILE：定义了一个或多个 SMT 的 YAML 或 JSON 文件的路径。
    • MESSAGE：示例消息的正文。
    • ATTRIBUTES：可选。以英文逗号分隔的消息属性列表。每个属性都是一个键值对，格式为 KEY="VALUE"。

    该命令按顺序执行 SMT，并将每个 SMT 的输出用作下一个 SMT 的输入。该命令会输出每个步骤的结果。

  5. 如需创建订阅，请运行 gcloud pubsub subscriptions create 命令：

    gcloud pubsub subscriptions create SUBSCRIPTION_ID \
    --topic=projects/PROJECT_ID/topics/TOPIC_ID \
    --message-transforms-file=TRANSFORMS_FILE

    替换以下内容：

    • SUBSCRIPTION_ID：您要创建的订阅的 ID 或名称。如需了解有关如何命名订阅的准则，请参阅资源名称。订阅的名称不可变。

    • PROJECT_ID：包含主题的项目的 ID。

    • TOPIC_ID：要订阅的主题的 ID。

    • TRANSFORMS_FILE：定义了一个或多个 SMT 的 YAML 或 JSON 文件的路径。

    6. Java

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

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

public class CreateSubscriptionWithSmtExample {
  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";

    createSubscriptionWithSmtExample(projectId, topicId, subscriptionId);
  }

  public static void createSubscriptionWithSmtExample(
      String projectId, String topicId, String subscriptionId) throws IOException {

    // UDF that removes the 'ssn' field, if present
    String code =
        "function redactSSN(message, metadata) {"
            + "  const data = JSON.parse(message.data);"
            + "  delete data['ssn'];"
            + "  message.data = JSON.stringify(data);"
            + "  return message;"
            + "}";
    String functionName = "redactSSN";

    JavaScriptUDF udf =
        JavaScriptUDF.newBuilder().setCode(code).setFunctionName(functionName).build();
    MessageTransform transform = MessageTransform.newBuilder().setJavascriptUdf(udf).build();

    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

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

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  // Add the UDF message transform
                  .addMessageTransforms(transform)
                  .build());

      System.out.println("Created subscription with SMT: " + subscription.getAllFields());
    }
  }
}

    Python

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

    from google.cloud import pubsub_v1
from google.pubsub_v1.types import JavaScriptUDF, MessageTransform

# TODO(developer): Choose an existing topic.
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"

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)

code = """function redactSSN(message, metadata) {
            const data = JSON.parse(message.data);
            delete data['ssn'];
            message.data = JSON.stringify(data);
            return message;
            }"""
udf = JavaScriptUDF(code=code, function_name="redactSSN")
transforms = [MessageTransform(javascript_udf=udf)]

with subscriber:
    subscription = subscriber.create_subscription(
        request={
            "name": subscription_path,
            "topic": topic_path,
            "message_transforms": transforms,
        }
    )
    print(f"Created subscription with SMT: {subscription}")

    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"
)

// createSubscriptionWithSMT creates a subscription with a single message transform function applied.
func createSubscriptionWithSMT(w io.Writer, projectID, topicID, subID string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	// subID := "my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	code := `function redactSSN(message, metadata) {
			const data = JSON.parse(message.data);
			delete data['ssn'];
			message.data = JSON.stringify(data);
			return message;
		}`

	transform := &pubsubpb.MessageTransform{
		Transform: &pubsubpb.MessageTransform_JavascriptUdf{
			JavascriptUdf: &pubsubpb.JavaScriptUDF{
				FunctionName: "redactSSN",
				Code:         code,
			},
		},
	}

	sub := &pubsubpb.Subscription{
		Name:              fmt.Sprintf("projects/%s/subscriptions/%s", projectID, subID),
		Topic:             fmt.Sprintf("projects/%s/topics/%s", projectID, topicID),
		MessageTransforms: []*pubsubpb.MessageTransform{transform},
	}
	sub, err = client.SubscriptionAdminClient.CreateSubscription(ctx, sub)
	if err != nil {
		return fmt.Errorf("CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created subscription with message transform: %v\n", sub)
	return nil
}

SMT 如何与其他订阅功能互动

如果您的订阅同时使用 SMT 和 Pub/Sub 的内置过滤条件，则系统会在应用 SMT 之前应用过滤条件。这具有以下含义：

  • 如果您的 SMT 更改了消息属性，Pub/Sub 过滤条件不会应用于新的属性集。
  • 您的 SMT 不会应用于被 Pub/Sub 过滤条件过滤掉的任何消息。

如果您的 SMT 过滤掉消息，请注意这对监控订阅积压的影响。 如果您将订阅提供给 Dataflow 流水线，请勿使用 SMT 过滤掉消息，因为这会中断 Dataflow 的自动扩缩。

后续步骤