使用 SMT 建立訂閱

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

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

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

事前準備

必要角色和權限

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

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

所需權限

如要使用 SMT 建立訂閱項目，必須具備下列權限：

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

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

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

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

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

建立含有 SMT 的訂閱項目

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

如要建立具有一或多個 SMT 的 Pub/Sub 訂閱項目，請執行下列步驟。

控制台

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

    前往「訂閱項目」頁面

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

    「建立訂閱」頁面隨即開啟。

  3. 在「Subscription ID」(訂閱 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，請按一下「刪除」。

  10. 選用。如要測試範例訊息的 SMT，請按照下列步驟操作：

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

    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. 選用。如要測試一或多個 SMT 是否能處理 Pub/Sub 訊息範例，請執行 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)。如果您仍在使用第 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 和 Pub/Sub 的內建篩選器，系統會先套用篩選器，再套用 SMT。這會造成下列影響：

  • 如果 SMT 會變更訊息屬性，Pub/Sub 篩選器就不會套用至新的屬性集。
  • 如果訊息遭到 Pub/Sub 篩選器篩除，系統就不會套用 SMT。

如果 SMT 篩除郵件，請注意這對監控訂閱項目待處理事項的影響。如果將訂閱項目饋送至 Dataflow 管道，請勿使用 SMT 篩除訊息，否則會中斷 Dataflow 的自動調度資源功能。

後續步驟