API 사용하기

API 사용하기

Pulsar 함수를 Java, Python, Go로 개발할 때 쓸 수 있는 API를 정리해 볼게요. 어떤 인터페이스를 고르느냐에 따라 필요한 라이브러리와 할 수 있는 일이 달라지니, 각각의 특징을 먼저 파악하는 게 중요해요.

크게 언어 네이티브 인터페이스, Pulsar 함수 SDK, 그리고 Java 전용 확장 SDK 세 가지로 나뉘어요. 함수가 컨텍스트(context)가 필요하냐, 외부 리소스를 초기화·해제해야 하냐에 따라 적절한 걸 선택하면 돼요.

출처: 문서

본문

아래 표는 Java, Python, Go에서 Pulsar 함수를 개발할 때 쓸 수 있는 API를 정리한 거예요.

Java/Python 언어 네이티브 인터페이스 사용하기 (Use language-native interface)

언어 네이티브 인터페이스는 Java/Python 함수를 작성하는 간단하고 깔끔한 방법이에요. 예를 들어 들어오는 모든 문자열에 느낌표를 붙이고 그 출력 문자열을 토픽에 게시하는 식이죠. 외부 의존성이 없어요.

아래 예시들은 언어 네이티브 함수예요.

Java:

Java 코드를 "언어 네이티브" 함수로 사용하려면 java.util.Function 인터페이스를 구현해야 해요. apply 메서드 안에 더 복잡한 처리 기능을 위한 어떤 로직이든 넣을 수 있어요.

import java.util.function.Function;

public class JavaNativeExclamationFunction implements Function<String, String> {
    @Override
    public String apply(String input) {
        return String.format("%s!", input);
    }
}

자세한 내용은 코드 예시를 참고해요.

Python:

Python 코드를 "언어 네이티브" 함수로 사용하려면 아래처럼 process라는 메서드가 있어야 해요. 이 메서드는 받은 문자열 값에 느낌표를 붙여요.

def process(input):
    return "{}!".format(input)

자세한 내용은 코드 예시를 참고해요.

알아두기: Pulsar 함수를 Python 3로 작성해요. 함수가 실행되도록 하려면 함수 워커에 Python 3가 설치되어 있고 기본 인터프리터로 설정되어 있어야 해요.

Java/Python/Go SDK 사용하기 (Use SDK)

Pulsar 함수 SDK의 구현은 컨텍스트 객체를 매개변수로 포함하는 함수형 인터페이스를 지정해요.

아래 예시들은 다양한 언어의 Pulsar 함수 SDK를 사용한 모습이에요.

Java:

Java SDK로 함수를 개발할 때는 org.apache.pulsar.functions.api.Function 인터페이스를 구현해야 해요. 이 인터페이스는 구현해야 할 메서드가 process 하나뿐이에요.

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

public class ExclamationFunction implements Function<String, String> {
    @Override
    public String process(String input, Context context) {
        return String.format("%s!", input);
    }
}

자세한 내용은 코드 예시를 참고해요.

함수의 반환 타입을 Record 제네릭으로 감쌀 수도 있는데, 이렇게 하면 토픽, 스키마, 프로퍼티 같은 출력 메시지를 더 세밀하게 제어할 수 있어요. Context::newOutputRecordBuilder 메서드를 사용해서 이 Record 출력을 만들면 돼요.

import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import org.apache.pulsar.functions.api.Record;

public class RecordFunction implements Function<String, Record<String>> {

    @Override
    public Record<String> process(String input, Context context) throws Exception {
        String output = String.format("%s!", input);
        Map<String, String> properties = new HashMap<>(context.getCurrentRecord().getProperties());
        context.getCurrentRecord().getTopicName().ifPresent(topic -> properties.put("input_topic", topic));

        return context.newOutputRecordBuilder(Schema.STRING)
                .value(output)
                .properties(properties)
                .build();
    }
}

자세한 내용은 코드 예시를 참고해요.

Python:

Python SDK로 함수를 개발하려면 Python 설치에 pulsar 클라이언트 의존성을 추가해야 해요.

from pulsar import Function

class ExclamationFunction(Function):
  def __init__(self):
    pass

  def process(self, input, context):
    return input + '!'

자세한 내용은 코드 예시를 참고해요.

Go:

Go SDK로 함수를 개발하려면 Go 설치에 pulsar 클라이언트 의존성을 추가하고, main() 메서드 안에서 pf.Start() 메서드에 함수 이름을 제공해야 해요. 이렇게 하면 함수가 Pulsar 함수 프레임워크에 등록되어 새 메시지가 도착했을 때 지정한 함수가 호출될 수 있어요.

package main

import (
	"context"
	"fmt"

	"github.com/apache/pulsar/pulsar-function-go/pf"
)

func HandleRequest(ctx context.Context, in []byte) error{
	fmt.Println(string(in) + "!")
	return nil
}

func main() {
	pf.Start(HandleRequest)
}

자세한 내용은 코드 예시를 참고해요.

Java 확장 SDK 사용하기 (Use extended SDK for Java)

이 확장 Pulsar 함수 SDK는 외부 리소스를 초기화하고 해제하는 두 가지 추가 인터페이스를 제공해요.

• initialize 인터페이스 — 함수 인스턴스가 시작할 때 한 번만 초기화하면 되는 외부 리소스를 초기화할 수 있어요.
• close 인터페이스 — 함수 인스턴스가 닫힐 때 참조하던 외부 리소스를 닫을 수 있어요.

알아두기: Java용 확장 Pulsar 함수 SDK는 Pulsar 2.10.0 이상 버전에서만 사용할 수 있어요. 사용하기 전에 Pulsar 2.10.0 이상 버전에서 함수 워커를 설정해야 해요.

아래 예시는 Java용 Pulsar 함수 SDK의 확장 인터페이스를 사용해서, 함수 인스턴스가 시작할 때 RedisClient를 초기화하고 함수 인스턴스가 닫힐 때 해제하는 모습을 보여줘요.

Java:

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import io.lettuce.core.RedisClient;

public class InitializableFunction implements Function<String, String> {
    private RedisClient redisClient;

    private void initRedisClient(Map<String, Object> connectInfo) {
        redisClient = RedisClient.create(connectInfo.get("redisURI"));
    }

    @Override
    public void initialize(Context context) {
        Map<String, Object> connectInfo = context.getUserConfigMap();
        redisClient = initRedisClient(connectInfo);
    }

    @Override
    public String process(String input, Context context) {
        String value = client.get(key);
        return String.format("%s-%s", input, value);
    }

    @Override
    public void close() {
        redisClient.close();
    }
}

더 알아보기 (Learn more)

• 함수를 처음 개발한다면 함수 개발 튜토리얼부터 시작해 보세요.
• 컨텍스트 객체가 제공하는 기능이 궁금하다면 함수 개발 API의 관련 문서를 확인해요.
• 개발한 함수를 배포하는 방법은 함수 배포 문서를 참고해요.