使用 SMT 建立訂閱

本文說明如何使用單一訊息轉換 (SMT) 建立 Pub/Sub 訂閱項目。

訂閱項目 SMT 可直接在 Pub/Sub 中簡單修改訊息資料與屬性。這項功能可在訊息傳送至訂閱端之前，進行資料清理、篩選或格式轉換。

如要建立含有 SMT 的訂閱項目，可以使用 Google Cloud 控制台、Google Cloud CLI、用戶端程式庫或 Pub/Sub API。

事前準備

• 瞭解 Pub/Sub 服務和相關術語。

• 瞭解 SMT。

必要角色和權限

如要取得建立含 SMT 的訂閱項目所需的權限，請要求管理員授予您專案的 Pub/Sub 編輯者 (roles/pubsub.editor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這個預先定義的角色具備使用 SMT 建立訂閱項目所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

必須具備下列權限，才能建立含有 SMT 的訂閱項目：

• 授予專案的建立訂閱項目權限：pubsub.subscriptions.create

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

視訂閱類型而定，您可能需要額外權限。 如要查看確切的權限清單，請參閱討論建立特定訂閱項目的文件。舉例來說，如果您要建立含有 SMT 的 BigQuery 訂閱項目，請參閱「建立 BigQuery 訂閱項目」。

如果您在與主題不同的專案中建立訂閱項目，則必須將 roles/pubsub.subscriber 角色授予包含訂閱項目的專案主體，且該主體位於包含主題的專案中。

您可以在專案層級和個別資源層級設定存取權控管。

建立含有 SMT 的訂閱項目

使用 SMT 建立訂閱項目前，請先詳閱「訂閱項目的屬性」說明文件。

注意： 強烈建議您為使用 SMT 的訂閱項目設定 dead-letter 主題。因 SMT 失敗而轉送至 dead-letter 主題的訊息，會包含具有 SMT 失敗原因的 CloudPubSubDeadLetterSourceSMTErrorMessage 屬性。

如要建立含有一個或多個 SMT 的 Pub/Sub 訂閱項目，請按照下列步驟操作。每個訂閱方案最多可啟用 5 個 SMT。

控制台

1. 前往 Google Cloud 控制台的 Pub/Sub「Subscriptions」(訂閱項目) 頁面。

   前往「訂閱項目」頁面

2. 按一下「Create Subscription」 (建立訂閱項目)。

3. 在「Subscription ID」(訂閱項目 ID) 欄位中，輸入訂閱項目的 ID。如要進一步瞭解如何命名訂閱項目，請參閱命名規範。

4. 在「轉換」下方，按一下「新增轉換」。

5. 選取「轉換類型」。如要進一步瞭解支援的 SMT 類型，請參閱「SMT 類型」。

6. 設定 SMT 的設定屬性。屬性集取決於 SMT 類型。詳情請參閱該 SMT 類型的說明文件。

7. 選用。如要驗證 SMT，請按一下「Validate」(驗證)。如果 SMT 有效，系統會顯示 "Validation passed" 訊息。否則系統會顯示錯誤訊息。

8. 如要新增其他轉換作業，請按一下「新增轉換作業」，然後重複上述步驟。

   如要依特定順序排列 SMT，請按一下 「Move up」(上移) 或 「Move down」(下移)。如要移除 SMT，請按一下「刪除」。

9. 選用。如要測試範例郵件的 SMT，請按照下列步驟操作：

   1. 按一下「測試轉換」。

   2. 在「Test transform」視窗中，選取要測試的函式。

   3. 在「輸入訊息」視窗中輸入範例訊息。

   4. 如要為訊息新增屬性，請按一下「新增屬性」，然後輸入屬性的鍵和值。你可以新增多項屬性。

   5. 按一下「測試」。對訊息套用 SMT 的結果會顯示在「Output message」(輸出訊息) 下方。

   6. 如要關閉「測試轉換」視窗，請按一下「關閉」圖示 。

   如果您建立多個 SMT，可以按照下列方式測試整個轉換序列：

   1. 如先前步驟所述，測試序列中的第一個 SMT。
   2. 選取下一個 SMT。輸入訊息會預先填入前一項測試的輸出訊息。
   3. 繼續依序測試 SMT，確保整個序列運作正常。

10. 按一下「建立」，建立訂閱項目。

gcloud

1. 在 Google Cloud 控制台中啟用 Cloud Shell。

   啟用 Cloud Shell

   Google Cloud 控制台底部會開啟 Cloud Shell 工作階段，並顯示指令列提示。Cloud Shell 是已安裝 Google Cloud CLI 的殼層環境，並已設定適用於您目前專案的值。工作階段可能要幾秒鐘的時間才能初始化。

2. 建立 YAML 或 JSON 檔案，定義一或多個 SMT。YAML 或 JSON 定義取決於 SMT 類型。詳情請參閱「SMT 類型」。

   如果檔案包含多個 SMT，Pub/Sub 會依列出的順序執行。

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：YAML 或 JSON 檔案的路徑，用於定義一或多個 SMT。

C#

在試用這個範例之前，請先按照「快速入門導覽課程：使用用戶端程式庫」中的 C# 設定操作說明進行操作。詳情請參閱 Pub/Sub C# API 參考文件。

using Google.Cloud.PubSub.V1;
using System;

public class CreateSubscriptionWithSingleMessageTransformSample
{
    public Subscription CreateSubscriptionWithSingleMessageTransform(string projectId, string topicId, string subscriptionId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);

        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
        MessageTransform removeSSNTransform = new MessageTransform
        {
            JavascriptUdf = new JavaScriptUDF
            {
                FunctionName = "redactSsn",
                Code = "function redactSsn(message, metadata) {"
                + "   const data = JSON.parse(message.data);"
                + "   delete data['ssn'];"
                + "   message.data = JSON.stringify(data);"
                + "   return message;"
                + "}"
            },
        };

        var subscription = subscriber.CreateSubscription(new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            MessageTransforms = { removeSSNTransform },
        });
        Console.WriteLine($"Subscription {subscription.Name} created with SMT.");

        return subscription;
    }
}

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)。如果您仍在使用第 1 版程式庫，請參閱第 2 版遷移指南。如要查看第 1 版程式碼範例清單，請參閱 已淘汰的程式碼範例。

在試用這個範例之前，請先按照「快速入門導覽課程：使用用戶端程式庫」中的 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 時，請注意下列事項。

篩選

如果訂閱項目同時使用 SMT 和 Pub/Sub 的內建篩選器，系統會先套用篩選器，再套用 SMT。這會產生下列影響：

• 如果 SMT 會變更訊息屬性，Pub/Sub 篩選器就不會套用至新的屬性集。
• SMT 不會套用至 Pub/Sub 篩選器篩除的訊息。
• 如果 SMT 篩除訊息，請注意這對監控訂閱積壓工作的影響。
• 如果將訂閱項目連結至 Dataflow 管道，請勿使用訂閱項目 SMT 篩除訊息，因為這會中斷 Dataflow 的自動調度資源功能。

訊息排序

如果您在已啟用排序功能的訂閱項目上定義 SMT，且執行 SMT 時擲回錯誤，則系統不會將具有相同排序鍵的後續訊息傳送至訂閱端。為避免這個問題，請在訂閱項目中設定無效信件主題，從訊息待處理項目中移除未處理的訊息。

後續步驟

• 排解訂閱 SMT 問題

• 更新訂閱 SMT

• 刪除訂閱項目 SMT