튜토리얼

함수를 처음부터 작성해서 배포하는 과정을 예제로 따라가 볼게요. 단어 개수 세기(Word Count), 내용 기반 라우팅, 윈도우 함수까지 세 가지 유형의 함수를 언어별로 작성해 볼 거예요. 예제를 하나씩 직접 만들다 보면 함수의 동작 방식이 자연스럽게 익숙해져요.

알아두기: 아래 예시들은 상태 저장 함수(stateful function)예요. 기본적으로 함수의 상태는 비활성화되어 있으니, 상태 저장 함수 활성화 방법을 먼저 확인해 두세요.

출처: 문서

본문

단어 개수 세는 함수 작성하기 (Write a function for word count)

단어 개수 세는 함수를 작성하려면 다음 단계를 완료해요.

Java SDK를 사용해 함수를 Java로 작성해요.

 package org.example.functions;

 import org.apache.pulsar.functions.api.Context;
 import org.apache.pulsar.functions.api.Function;

 import java.util.Arrays;

 public class WordCountFunction implements Function<String, Void> {
     // This function is invoked every time a message is published to the input topic
     @Override
     public Void process(String input, Context context) throws Exception {
         Arrays.asList(input.split(" ")).forEach(word -> {
             String counterKey = word.toLowerCase();
             context.incrCounter(counterKey, 1);
         });
         return null;
     }
 }

JAR 파일을 번들로 빌드한 다음, pulsar-admin 명령으로 Pulsar 클러스터에 배포해요.

bin/pulsar-admin functions create \
   --jar $PWD/target/my-jar-with-dependencies.jar \
   --classname org.example.functions.WordCountFunction \
   --tenant public \
   --namespace default \
   --name word-count \
   --inputs persistent://public/default/sentences \
   --output persistent://public/default/count

내용 기반 라우팅 함수 작성하기 (Write a function for content-based routing)

내용 기반 라우팅 함수를 작성하려면 다음 단계를 완료해요.

Python SDK를 사용해 함수를 Python으로 작성해요.

 from pulsar import Function

 class RoutingFunction(Function):
     def __init__(self):
         self.fruits_topic = "persistent://public/default/fruits"
         self.vegetables_topic = "persistent://public/default/vegetables"

     def is_fruit(item):
         return item in [b"apple", b"orange", b"pear", b"other fruits..."]

     def is_vegetable(item):
         return item in [b"carrot", b"lettuce", b"radish", b"other vegetables..."]

     def process(self, item, context):
         if self.is_fruit(item):
             context.publish(self.fruits_topic, item)
         elif self.is_vegetable(item):
             context.publish(self.vegetables_topic, item)
         else:
             warning = "The item {0} is neither a fruit nor a vegetable".format(item)
             context.get_logger().warn(warning)

이 코드를 ~/router.py에 저장했다고 가정하면, pulsar-admin 명령으로 Pulsar 클러스터에 배포할 수 있어요.

bin/pulsar-admin functions create \
   --py ~/router.py \
   --classname router.RoutingFunction \
   --tenant public \
   --namespace default \
   --name route-fruit-veg \
   --inputs persistent://public/default/basket-items

단어 개수 세는 윈도우 함수 작성하기 (Write a window function for word count)

알아두기: 현재 윈도우 함수는 Java에서만 사용할 수 있어요.

이 예시는 언어 네이티브 인터페이스를 사용해 Java로 윈도우 함수를 작성하는 방법을 보여줘요.

각 입력 메시지는 문장이고, 단어로 분리된 다음 각 단어가 세어져요. 내장 카운터 상태는 단어 개수를 지속적(persistent)이고 일관성 있게 유지하는 데 사용돼요.

public class WordCountFunction implements Function<String, Void> {
    @Override
    public Void process(String input, Context context) {
        Arrays.asList(input.split("\\s+")).forEach(word -> context.incrCounter(word, 1));
        return null;
    }
}

더 알아보기 (Learn more)

함수의 상태 저장을 활성화하는 방법은 상태 저장 함수 문서를 참고해요.
함수 배포 명령의 상세 옵션은 함수 배포 문서에서 확인할 수 있어요.
함수 개발의 기초 개념이 궁금하다면 함수 개발 API 문서를 살펴보세요.