SMT로 구독 만들기

이 문서에서는 단일 메시지 변환 (SMT)으로 Pub/Sub 구독을 만드는 방법을 설명합니다.

구독 SMT를 사용하면 Pub/Sub 내에서 직접 메시지 데이터와 속성을 가볍게 수정할 수 있습니다. 이 기능을 사용하면 메시지가 구독자 클라이언트에 전달되기 전에 데이터 정리, 필터링 또는 형식 변환을 수행할 수 있습니다.

SMT로 구독을 만들려면 Google Cloud 콘솔, Google Cloud CLI, 클라이언트 라이브러리 또는 Pub/Sub API를 사용하면 됩니다.

시작하기 전에

필수 역할 및 권한

SMT로 구독을 만드는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 Pub/Sub 편집자 (roles/pubsub.editor) IAM 역할을 부여해 달라고 요청하세요. 역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

이 사전 정의된 역할에는 SMT로 구독을 만드는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.

필수 권한

SMT로 구독을 만들려면 다음 권한이 필요합니다.

  • 프로젝트에 구독 만들기 권한 부여: pubsub.subscriptions.create

커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한을 부여받을 수도 있습니다.

구독 유형에 따라 추가 권한이 필요할 수 있습니다. 정확한 권한 목록을 확인하려면 특정 구독 만들기를 설명하는 문서를 참조하세요. 예를 들어 SMT로 BigQuery 구독을 만드는 경우 BigQuery 구독 만들기를 참조하세요.

주제와 다른 프로젝트에서 구독을 만드는 경우 주제가 포함된 프로젝트에서 구독이 포함된 프로젝트의 보안 주체에 roles/pubsub.subscriber 역할을 부여해야 합니다.

프로젝트 수준 및 개별 리소스 수준에서 액세스 제어를 구성할 수 있습니다.

SMT로 구독 만들기

SMT로 구독을 만들기 전에 구독 속성 문서를 검토하세요.

하나 이상의 SMT로 Pub/Sub 구독을 만들려면 다음 단계를 수행하세요. 구독당 최대 5개의 SMT를 사용 설정할 수 있습니다.

콘솔

  1. 콘솔에서 Pub/Sub 구독 페이지로 이동합니다. Google Cloud

    구독 페이지로 이동

  2. 구독 만들기 를 클릭합니다.

  3. 구독 ID 필드에 구독 ID를 입력합니다. 구독 이름 지정에 대한 자세한 내용은 이름 지정 가이드라인을 참조하세요.

  4. **변환** 에서 **변환 추가** 를 클릭합니다.

  5. 변환 유형 을 선택합니다. 지원되는 SMT 유형에 대한 자세한 내용은 SMT 유형을 참조하세요.

  6. SMT의 구성 속성을 설정합니다. 속성 집합은 SMT 유형에 따라 다릅니다. 자세한 내용은 해당 SMT 유형의 문서를 참조하세요.

  7. 선택사항입니다. SMT를 검사하려면 검사 를 클릭합니다. SMT가 유효하면 메시지 "Validation passed"가 표시됩니다. 그렇지 않으면 오류 메시지가 표시됩니다.

  8. 다른 변환을 추가하려면 변환 추가 를 클릭하고 이전 단계를 반복합니다.

    SMT를 특정 순서로 정렬하려면 위로 이동 또는 아래로 이동 을 클릭합니다. SMT를 삭제하려면 삭제를 클릭합니다.

  9. 선택사항입니다. 샘플 메시지에서 SMT를 테스트하려면 다음 단계를 수행하세요.

    1. 변환 테스트 를 클릭합니다.

    2. 변환 테스트 창에서 테스트할 함수를 선택합니다.

    3. 입력 메시지 창에 샘플 메시지를 입력합니다.

    4. 메시지에 속성을 추가하려면 속성 추가 를 클릭하고 속성의 키와 값을 입력합니다. 여러 속성을 추가할 수 있습니다.

    5. 테스트 를 클릭합니다. 메시지에 SMT를 적용한 결과가 출력 메시지 아래에 표시됩니다.

    6. 변환 테스트 창을 닫으려면 닫기를 클릭합니다.

    SMT를 두 개 이상 만드는 경우 다음과 같이 전체 변환 시퀀스를 테스트할 수 있습니다.

    1. 이전 단계에 설명된 대로 시퀀스의 첫 번째 SMT를 테스트합니다.
    2. 다음 SMT를 선택합니다. 입력 메시지는 이전 테스트의 출력 메시지로 자동 입력됩니다.
    3. 전체 시퀀스가 예상대로 작동하는지 확인하기 위해 SMT를 순서대로 계속 테스트합니다.

  10. 만들기 를 클릭하여 구독을 만듭니다.

gcloud

  1. 콘솔에서 Cloud Shell을 활성화합니다. Google Cloud

    Cloud Shell 활성화

    콘솔 하단에 Cloud Shell 세션이 시작되고 명령줄 프롬프트가 표시됩니다. Google Cloud Cloud Shell은 Google Cloud CLI가 사전 설치된 셸 환경으로, 현재 프로젝트의 값이 이미 설정되어 있습니다. 세션이 초기화되는 데 몇 초 정도 걸릴 수 있습니다.

  2. 하나 이상의 SMT를 정의하는 YAML 또는 JSON 파일을 만듭니다. 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: 하나 이상의 SMT를 정의하는 YAML 또는 JSON 파일의 경로입니다.

자바

이 샘플을 시도하기 전에 빠른 시작: 클라이언트 라이브러리 사용의 Java 설정 안내를 따르세요. 자세한 내용은 Pub/Sub 자바 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를 사용할 때는 다음 사항을 고려하세요.

필터링

구독에서 SMT와 Pub/Sub의 기본 제공 필터를 모두 사용하는 경우 필터 가 SMT 전에 적용됩니다. 이는 다음과 같은 영향을 미칩니다.

  • SMT가 메시지 속성을 변경하는 경우 Pub/Sub 필터가 새 속성 집합에 적용되지 않습니다.
  • SMT는 Pub/Sub 필터로 필터링된 메시지에 적용되지 않습니다.
  • SMT가 메시지를 필터링하는 경우 구독 백로그 모니터링에 미치는 영향을 알아야 합니다.
  • 구독을 Dataflow 파이프라인에 연결하는 경우 구독 SMT를 사용하여 메시지를 필터링하지 마세요. Dataflow의 자동 확장을 방해하기 때문입니다.

메시지 순서

순서가 사용 설정된 구독에 SMT를 정의하고 SMT를 실행할 때 오류가 발생하면 동일한 순서 키의 후속 메시지가 구독자에게 전달되지 않습니다. 이 문제를 방지하려면 구독에 데드 레터 주제를 설정하여 처리되지 않은 메시지를 메시지 백로그에서 삭제하세요.

다음 단계