Pulsar Functions 시작하기

이 실습 튜토리얼은 standalone Pulsar에서 함수를 만들고 검증하는 방법을 단계별 예시로 보여줘요. 상태 저장 함수(stateful function)와 윈도우 함수(window function)도 함께 다뤄요.

함수를 처음 시작하는 분이라면 이 가이드를 처음부터 끝까지 따라 하면서 개념과 명령어를 함께 익히는 걸 추천해요. 각 출력값을 직접 눈으로 확인하면서 진행하면 훨씬 이해가 잘 돼요.

출처: 문서

본문

이 실습 튜토리얼에서는 standalone Pulsar에서 상태 저장 함수와 윈도우 함수를 포함해 함수를 만들고 검증하는 방법을 단계별 예시와 함께 보여줘요.

사전 준비 (Prerequisites)

JDK 8 이상. 자세한 내용은 Pulsar 런타임 Java 버전 권장 사항을 참고해요.
Windows OS는 지원되지 않아요.

1단계: standalone Pulsar 시작하기 (Start standalone Pulsar)

conf/standalone.conf에서 pulsar 함수를 활성화해요 (없으면 이 필드를 추가해요).

functionsWorkerEnabled=true

로컬에서 Pulsar를 시작해요.

bin/pulsar standalone

Pulsar 서비스의 모든 구성 요소(ZooKeeper, BookKeeper, broker 등)가 순서대로 시작돼요. bin/pulsar-admin brokers healthcheck 명령으로 Pulsar 서비스가 실행 중인지 확인할 수 있어요.

Pulsar 이진 프로토콜 포트를 확인해요.

telnet localhost 6650

Pulsar Function 클러스터를 확인해요.

bin/pulsar-admin functions-worker get-cluster

출력:

[{"workerId":"c-standalone-fw-localhost-6750","workerHostname":"localhost","port":6750}]

public 테넌트가 존재하는지 확인해요.

bin/pulsar-admin tenants list

출력:

public

default 네임스페이스가 존재하는지 확인해요.

bin/pulsar-admin namespaces list public

출력:

public/default

테이블 서비스(table service)가 성공적으로 활성화되었는지 확인해요.

telnet localhost 4181

출력:

Trying ::1...
telnet: connect to address ::1: Connection refused
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.

2단계: 테스트용 네임스페이스 만들기 (Create a namespace for test)

테넌트와 네임스페이스를 만들어요.

bin/pulsar-admin tenants create test
bin/pulsar-admin namespaces create test/test-namespace

1단계와 같은 터미널 창에서 테넌트와 네임스페이스를 확인해요.

bin/pulsar-admin namespaces list test

출력:

"test/test-namespace"가 출력되면 테넌트와 네임스페이스가 모두 성공적으로 생성된 거예요.

3단계: 함수 시작하기 (Start functions)

알아두기: 함수를 시작하기 전에 Pulsar를 시작하고 테스트 네임스페이스를 만들어야 해요.

examples라는 함수를 만들어요.

팁: 로컬 머신의 Pulsar 디렉터리 examples 폴더 아래에 example-function-config.yaml과 api-examples.jar 파일이 있는 걸 확인할 수 있어요.

이 예시 함수는 모든 메시지 끝에 !를 추가해요.

bin/pulsar-admin functions create \
   --function-config-file $PWD/examples/example-function-config.yaml \
   --jar $PWD/examples/api-examples.jar

출력:

Created Successfully

이 함수의 설정은 examples/example-function-config.yaml에서 확인할 수 있어요.

tenant: "test"
namespace: "test-namespace"
name: "example" # function name
className: "org.apache.pulsar.functions.api.examples.ExclamationFunction"
inputs: ["test_src"] # this function will read messages from these topics
output: "test_result" # the return value of this function will be sent to this topic
autoAck: true # function will acknowledge input messages if set true
parallelism: 1

ExclamationFunction의 소스 코드는 여기에서 확인할 수 있어요. yaml 설정에 대한 자세한 내용은 레퍼런스를 참고해요.

1단계와 같은 터미널 창에서 함수의 구성을 확인해요.

bin/pulsar-admin functions get \
   --tenant test \
   --namespace test-namespace \
   --name example

출력:

{
    "tenant": "test",
    "namespace": "test-namespace",
    "name": "example",
    "className": "org.apache.pulsar.functions.api.examples.ExclamationFunction",
    "inputSpecs": {
        "test_src": {
        "isRegexPattern": false,
        "schemaProperties": {},
        "consumerProperties": {},
        "poolMessages": false
        }
    },
    "output": "test_result",
    "producerConfig": {
        "useThreadLocalProducers": false,
        "batchBuilder": ""
    },
    "processingGuarantees": "ATLEAST_ONCE",
    "retainOrdering": false,
    "retainKeyOrdering": false,
    "forwardSourceMessageProperty": true,
    "userConfig": {},
    "runtime": "JAVA",
    "autoAck": true,
    "parallelism": 1,
    "resources": {
        "cpu": 1.0,
        "ram": 1073741824,
        "disk": 10737418240
    },
    "cleanupSubscription": true,
    "subscriptionPosition": "Latest"
}

1단계와 같은 터미널 창에서 함수의 상태를 확인해요.

bin/pulsar-admin functions status \
   --tenant test \
   --namespace test-namespace \
   --name example

출력:

"running": true는 함수가 실행 중임을 나타내요.

{
  "numInstances" : 1,
  "numRunning" : 1,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : true,
      "error" : "",
      "numRestarts" : 0,
      "numReceived" : 0,
      "numSuccessfullyProcessed" : 0,
      "numUserExceptions" : 0,
      "latestUserExceptions" : [ ],
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "averageLatency" : 0.0,
      "lastInvocationTime" : 0,
      "workerId" : "c-standalone-fw-localhost-8080"
    }
  } ]
}

1단계와 같은 터미널 창에서 출력 토픽 test_result를 구독해요.

bin/pulsar-client consume -s test-sub -n 0 test_result

새 터미널 창에서 입력 토픽 test_src에 메시지를 생산해요.

bin/pulsar-client produce -m "test-messages-`date`" -n 10 test_src

1단계와 같은 터미널 창에서 예시 함수가 생산한 메시지가 반환돼요. 모든 메시지 끝에 !가 추가된 것을 확인할 수 있어요.

출력:

----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!
----- got message -----
test-messages-Thu Jul 19 11:59:15 PDT 2021!

상태 저장 함수 시작하기 (Start stateful functions)

Pulsar의 standalone 모드는 상태 저장 함수를 위한 BookKeeper 테이블 서비스를 활성화해요. 자세한 내용은 상태 저장 구성 문서를 참고해요.

알아두기: 상태 저장 함수를 시작하기 전에 Pulsar를 시작하고 테스트 네임스페이스를 만들어야 해요.

다음 예시는 카운터 함수를 검증할 수 있는 상태 저장 함수를 시작하는 방법을 안내해요.

examples/example-stateful-function-config.yaml을 사용해 함수를 만들어요.

bin/pulsar-admin functions create \
   --function-config-file $PWD/examples/example-stateful-function-config.yaml \
   --jar $PWD/examples/api-examples.jar

출력:

Created Successfully

이 함수의 설정은 examples/example-stateful-function-config.yaml에서 확인할 수 있어요.

tenant: "test"
namespace: "test-namespace"
name: "word_count"
className: "org.apache.pulsar.functions.api.examples.WordCountFunction"
inputs: ["test_wordcount_src"] # this function will read messages from these topics
autoAck: true
parallelism: 1

WordCountFunction의 소스 코드는 여기에서 확인할 수 있어요. 이 함수는 어떤 값도 반환하지 않고 함수 컨텍스트에 단어의 출현 횟수를 저장해요. 그래서 출력 토픽을 지정할 필요가 없어요. yaml 설정에 대한 자세한 내용은 레퍼런스를 참고해요.

1단계와 같은 터미널 창에서 word_count 함수의 정보를 가져와요.

bin/pulsar-admin functions get \
   --tenant test \
   --namespace test-namespace \
   --name word_count

출력:

{
  "tenant": "test",
  "namespace": "test-namespace",
  "name": "word_count",
  "className": "org.apache.pulsar.functions.api.examples.WordCountFunction",
  "inputSpecs": {
    "test_wordcount_src": {
      "isRegexPattern": false,
      "schemaProperties": {},
      "consumerProperties": {},
      "poolMessages": false
    }
  },
  "producerConfig": {
    "useThreadLocalProducers": false,
    "batchBuilder": ""
  },
  "processingGuarantees": "ATLEAST_ONCE",
  "retainOrdering": false,
  "retainKeyOrdering": false,
  "forwardSourceMessageProperty": true,
  "userConfig": {},
  "runtime": "JAVA",
  "autoAck": true,
  "parallelism": 1,
  "resources": {
    "cpu": 1.0,
    "ram": 1073741824,
    "disk": 10737418240
  },
  "cleanupSubscription": true,
  "subscriptionPosition": "Latest"
}

1단계와 같은 터미널 창에서 word_count 함수의 상태를 가져와요.

bin/pulsar-admin functions status \
   --tenant test \
   --namespace test-namespace \
   --name word_count

출력:

{
  "numInstances" : 1,
  "numRunning" : 1,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : true,
       "error" : "",
       "numRestarts" : 0,
       "numReceived" : 0,
       "numSuccessfullyProcessed" : 0,
       "numUserExceptions" : 0,
       "latestUserExceptions" : [ ],
       "numSystemExceptions" : 0,
       "latestSystemExceptions" : [ ],
       "averageLatency" : 0.0,
       "lastInvocationTime" : 0,
       "workerId" : "c-standalone-fw-localhost-8080"
    }
  } ]
}

1단계와 같은 터미널 창에서 hello 키로 함수의 상태 테이블을 조회해요. 이 작업은 hello와 관련된 변경 사항을 관찰(watch)해요.

bin/pulsar-admin functions querystate \
   --tenant test \
   --namespace test-namespace \
   --name word_count -k hello -w

팁: pulsar-admin functions querystate 옵션 명령(플래그, 설명, 기본값, 단축형 포함)에 대한 자세한 내용은 Pulsar admin API를 참고해요.

출력:

key 'hello' doesn't exist.
key 'hello' doesn't exist.
key 'hello' doesn't exist.
...

새 터미널 창에서 다음 방법 중 하나로 입력 토픽 test_wordcount_src에 hello 메시지를 10개 생산해요. hello의 값이 10으로 갱신돼요.

bin/pulsar-client produce -m "hello" -n 10 test_wordcount_src

1단계와 같은 터미널 창에서 결과를 확인해요.

결과는 출력 토픽 test_wordcount_dest가 메시지를 받았음을 보여줘요.

출력:

{
  "key": "hello",
  "numberValue": 10,
  "version": 9
}

5단계 터미널 창에서 hello 메시지를 다시 10개 생산해요. hello의 값이 20으로 갱신돼요.

bin/pulsar-client produce -m "hello" -n 10 test_wordcount_src

1단계와 같은 터미널 창에서 결과를 확인해요.

결과는 출력 토픽 test_wordcount_dest가 값 20을 받았음을 보여줘요.

{
   "key": "hello",
   "numberValue": 20,
   "version": 19
}

윈도우 함수 시작하기 (Start window functions)

윈도우 함수는 Pulsar 함수의 특별한 형태예요. 자세한 내용은 개념 문서를 참고해요.

알아두기: 윈도우 함수를 시작하기 전에 Pulsar를 시작하고 테스트 네임스페이스를 만들어야 해요.

다음 예시는 윈도우 안의 합계를 계산하는 윈도우 함수를 시작하는 방법을 안내해요.

example-window-function-config.yaml을 사용해 함수를 만들어요.

bin/pulsar-admin functions create \
   --function-config-file $PWD/examples/example-window-function-config.yaml \
   --jar $PWD/examples/api-examples.jar

출력:

Created Successfully

이 함수의 설정은 examples/example-window-function-config.yaml에서 확인할 수 있어요.

tenant: "test"
namespace: "test-namespace"
name: "window-example"
className: "org.apache.pulsar.functions.api.examples.AddWindowFunction"
inputs: ["test_window_src"]
output: "test_window_result"
autoAck: true
parallelism: 1

# every 5 messages, calculate sum of the latest 10 messages
windowConfig:
  windowLengthCount: 10
  slidingIntervalCount: 5

AddWindowFunction의 소스 코드는 여기에서 확인할 수 있어요. yaml 설정에 대한 자세한 내용은 레퍼런스를 참고해요.

1단계와 같은 터미널 창에서 함수의 구성을 확인해요.

bin/pulsar-admin functions get \
   --tenant test \
   --namespace test-namespace \
   --name window-example

출력:

{
  "tenant": "test",
  "namespace": "test-namespace",
  "name": "window-example",
  "className": "org.apache.pulsar.functions.api.examples.AddWindowFunction",
  "inputSpecs": {
    "test_window_src": {
      "isRegexPattern": false,
      "schemaProperties": {},
      "consumerProperties": {},
      "poolMessages": false
    }
  },
  "output": "test_window_result",
  "producerConfig": {
    "useThreadLocalProducers": false,
    "batchBuilder": ""
  },
  "processingGuarantees": "ATLEAST_ONCE",
  "retainOrdering": false,
  "retainKeyOrdering": false,
  "forwardSourceMessageProperty": true,
  "userConfig": {},
  "runtime": "JAVA",
  "autoAck": false,
  "parallelism": 1,
  "resources": {
    "cpu": 1.0,
    "ram": 1073741824,
    "disk": 10737418240
  },
  "windowConfig": {
    "windowLengthCount": 10,
    "slidingIntervalCount": 5,
    "actualWindowFunctionClassName": "org.apache.pulsar.functions.api.examples.AddWindowFunction",
    "processingGuarantees": "ATLEAST_ONCE"
  },
  "cleanupSubscription": true,
  "subscriptionPosition": "Latest"
}

1단계와 같은 터미널 창에서 함수의 상태를 확인해요.

bin/pulsar-admin functions status \
   --tenant test \
   --namespace test-namespace \
   --name window-example

출력:

"running": true는 함수가 실행 중임을 나타내요.

{
  "numInstances" : 1,
  "numRunning" : 1,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : true,
      "error" : "",
      "numRestarts" : 0,
      "numReceived" : 0,
      "numSuccessfullyProcessed" : 0,
      "numUserExceptions" : 0,
      "latestUserExceptions" : [ ],
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "averageLatency" : 0.0,
      "lastInvocationTime" : 0,
      "workerId" : "c-standalone-fw-localhost-8080"
    }
  } ]
}

1단계와 같은 터미널 창에서 출력 토픽 test_window_result를 구독해요.

bin/pulsar-client consume -s test-sub -n 0 test_window_result

새 터미널 창에서 입력 토픽 test_window_src에 메시지를 생산해요.

bin/pulsar-client produce -m "3" -n 10 test_window_src

1단계와 같은 터미널 창에서 윈도우 함수 window-example이 생산한 메시지가 반환돼요.

출력:

----- got message -----
key:[null], properties:[], content:15
----- got message -----
key:[null], properties:[], content:30

더 알아보기 (Learn more)

함수를 직접 개발하는 방법이 궁금하다면 함수 개발 튜토리얼 문서를 참고해요.
함수의 개념이 궁금하다면 Pulsar 함수 개요 문서를 살펴보세요.
함수 관리 명령어 전체는 Pulsar admin API 문서에서 확인할 수 있어요.