상태 저장 구성하기

상태 저장 구성하기

Pulsar 함수는 Apache BookKeeper를 상태 저장(state storage) 인터페이스로 사용해요. Pulsar는 BookKeeper 테이블 서비스와 통합해서 함수의 상태를 저장하는데, 예를 들어 WordCount 함수는 State API를 통해 카운터 상태를 BookKeeper 테이블 서비스에 저장할 수 있어요.

상태는 키-값 쌍이에요. 키는 문자열이고 값은 임의의 이진 데이터이죠. 카운터는 64비트 big-endian 이진 값으로 저장돼요. 키는 개별 함수에 한정(scoped)되고, 그 함수의 인스턴스들 사이에서 공유돼요.

알아두기: Go 함수에는 상태 저장을 사용할 수 없어요.

출처: 문서

본문

State API 호출하기 (Call state APIs)

Pulsar 함수는 상태를 변경하고 접근하는 API를 제공해요. Java/Python SDK로 함수를 개발할 때 이 API들은 Context 객체에서 사용할 수 있어요.

아래 표는 Java와 Python 함수 안에서 접근할 수 있는 상태 관련 API를 정리한 거예요.

상태 관련 API	Java	Python
카운터 증가	incrCounter, incrCounterAsync	incr_counter
카운터 조회	getCounter, getCounterAsync	get_counter
상태 갱신	putState, putStateAsync	put_state
상태 조회	getState, getStateAsync	get_state
상태 삭제	deleteState	del_counter

카운터 증가 (Increment counter)

incrCounter를 사용하면 주어진 키의 카운터를 주어진 양만큼 증가시킬 수 있어요. 키가 존재하지 않으면 새 키가 생성돼요.

Java:

    /**
     * Increment the built-in distributed counter referred by key
     * @param key The name of the key
     * @param amount The amount to be incremented
     */
    void incrCounter(String key, long amount);

카운터를 비동기로 증가시키려면 incrCounterAsync를 사용하면 돼요.

    /**
     * Increment the built-in distributed counter referred by key
     * but dont wait for the completion of the increment operation
     *
     * @param key The name of the key
     * @param amount The amount to be incremented
     */
    CompletableFuture<Void> incrCounterAsync(String key, long amount);

Python:

  def incr_counter(self, key, amount):
    """incr the counter of a given key in the managed state"""

카운터 조회 (Retrieve counter)

getCounter를 사용하면 incrCounter로 변경된 주어진 키의 카운터를 조회할 수 있어요.

Java:

    /**
     * Retrieve the counter value for the key.
     *
     * @param key name of the key
     * @return the amount of the counter value for this key
     */
    long getCounter(String key);

incrCounterAsync로 변경된 카운터를 비동기로 조회하려면 getCounterAsync를 사용해요.

    /**
     * Retrieve the counter value for the key, but don't wait
     * for the operation to be completed
     *
     * @param key name of the key
     * @return the amount of the counter value for this key
     */
    CompletableFuture<Long> getCounterAsync(String key);

Python:

  def get_counter(self, key):
    """get the counter of a given key in the managed state"""

상태 갱신 (Update state)

카운터 API 외에도 Pulsar는 함수가 주어진 키의 상태를 저장하고 갱신할 수 있게 범용 키/값 API를 제공해요.

Java:

    /**
     * Update the state value for the key.
     *
     * @param key name of the key
     * @param value state value of the key
     */
    void putState(String key, ByteBuffer value);

주어진 키의 상태를 비동기로 갱신하려면 putStateAsync를 사용해요.

    /**
     * Update the state value for the key, but don't wait for the operation to be completed
     *
     * @param key name of the key
     * @param value state value of the key
     */
    CompletableFuture<Void> putStateAsync(String key, ByteBuffer value);

Python:

  def put_state(self, key, value):
    """update the value of a given key in the managed state"""

상태 조회 (Retrieve state)

getState를 사용하면 주어진 키의 상태를 조회할 수 있어요.

Java:

    /**
     * Retrieve the state value for the key.
     *
     * @param key name of the key
     * @return the state value for the key.
     */
    ByteBuffer getState(String key);

주어진 키의 상태를 비동기로 조회하려면 getStateAsync를 사용해요.

    /**
     * Retrieve the state value for the key, but don't wait for the operation to be completed
     *
     * @param key name of the key
     * @return the state value for the key.
     */
    CompletableFuture<ByteBuffer> getStateAsync(String key);

Python:

  def get_state(self, key):
    """get the value of a given key in the managed state"""

상태 삭제 (Delete state)

알아두기: 카운터와 이진 값 모두 같은 키 공간(keyspace)을 공유하므로, 이 API는 두 종류 모두를 삭제해요.

Java:

    /**
     * Delete the state value for the key.
     *
     * @param key   name of the key
     */
    void deleteState(String key);

CLI로 상태 조회하기 (Query state via CLI)

State API로 함수의 상태를 Pulsar 상태 저장소에 저장하고 다시 꺼내 쓰는 것 외에도, CLI 명령으로 함수의 상태를 조회할 수 있어요.

bin/pulsar-admin functions querystate \
    --tenant <tenant> \
    --namespace <namespace> \
    --name <function-name> \
    --state-storage-url <bookkeeper-service-url> \
    --key <state-key> \
    [---watch]

--watch를 지정하면 CLI 도구가 계속 실행되면서 주어진 state-key의 최신 값을 가져와요.

예시 (Example)

WordCountFunction 예시는 Pulsar 함수 안에서 상태가 어떻게 저장되는지 보여줘요.

Java:

WordCountFunction의 로직은 단순하고 직관적이에요.

• 함수는 정규식 \.을 사용해 수신한 문자열을 여러 단어로 분리해요.
• 각 단어에 대해 incrCounter(key, amount)로 카운터를 1씩 증가시켜요.

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> {
    @Override
    public Void process(String input, Context context) throws Exception {
        Arrays.asList(input.split("\\.")).forEach(word -> context.incrCounter(word, 1));
        return null;
    }
}

Python:

이 WordCount 함수의 로직도 단순하고 직관적이에요.

• 함수는 먼저 수신한 문자열을 여러 단어로 분리해요.
• 각 단어에 대해 incr_counter(key, amount)로 카운터를 1씩 증가시켜요.

from pulsar import Function

class WordCount(Function):
    def process(self, item, context):
        for word in item.split():
            context.incr_counter(word, 1)

더 알아보기 (Learn more)

• 상태 저장을 사용하려면 함수의 상태를 활성화해야 해요. 상태 저장 함수 문서를 참고해요.
• 상태 저장의 기반이 되는 BookKeeper 테이블 서비스가 궁금하다면 BookKeeper 문서를 살펴보세요.
• 함수 개발에 필요한 다른 기능은 함수 개발 API 문서에서 확인할 수 있어요.