윈도우 함수 컨텍스트

윈도우 함수 컨텍스트 (Window functions context)

Java SDK는 윈도우 함수가 사용할 수 있는 윈도우 컨텍스트(window context) 객체에 접근할 수 있게 해줘요. 이 컨텍스트 객체는 Pulsar 윈도우 함수를 위한 다양한 정보와 기능을 제공해요. Spec 같은 함수 기본 정보부터 로거, 사용자 설정, 라우팅, 메트릭, 상태 저장까지 컨텍스트로 접근할 수 있으니 하나씩 살펴볼게요.

출처: 문서

본문

Java SDK는 윈도우 함수가 사용할 수 있는 윈도우 컨텍스트 객체에 접근을 제공해요. 이 컨텍스트 객체는 Pulsar 윈도우 함수를 위한 다양한 정보와 기능을 아래와 같이 제공해요.

• Spec: 함수와 연결된 모든 입력 토픽과 출력 토픽의 이름. 함수와 연결된 테넌트(tenant)와 네임스페이스(namespace). Pulsar 윈도우 함수의 이름, ID, 버전. 윈도우 함수를 실행하는 Pulsar 함수 인스턴스의 ID. 윈도우 함수를 호출하는 인스턴스의 수. 출력 스키마의 내장형 타입 또는 사용자 정의 클래스 이름.
• Logger: 윈도우 함수가 사용하는 Logger 객체로, 윈도우 함수 로그 메시지를 만드는 데 쓸 수 있어요.
• User config: 임의의 사용자 설정 값에 접근할 수 있어요.
• Routing: Pulsar 윈도우 함수에서 라우팅을 지원해요. Pulsar 윈도우 함수는 publish 인터페이스에 따라 임의의 토픽으로 메시지를 보내요.
• Metrics: 메트릭을 기록하는 인터페이스.
• State storage: 상태 저장소에 상태를 저장하고 검색하는 인터페이스.

Spec

Spec은 함수의 기본 정보를 담고 있어요.

입력 토픽 가져오기 (Get input topics)

getInputTopics 메서드는 모든 입력 토픽의 이름 목록을 가져와요. 이 예시는 Java 윈도우 함수에서 모든 입력 토픽의 이름 목록을 가져오는 방법을 보여줘요.

public class GetInputTopicsWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        Collection<String> inputTopics = context.getInputTopics();
        System.out.println(inputTopics);

        return null;
    }

}

출력 토픽 가져오기 (Get output topic)

getOutputTopic 메서드는 메시지가 전송되는 토픽의 이름을 가져와요. 이 예시는 Java 윈도우 함수에서 출력 토픽의 이름을 가져오는 방법을 보여줘요.

public class GetOutputTopicWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        String outputTopic = context.getOutputTopic();
        System.out.println(outputTopic);

        return null;
    }
}

테넌트 가져오기 (Get tenant)

getTenant 메서드는 윈도우 함수와 연결된 테넌트 이름을 가져와요. 이 예시는 Java 윈도우 함수에서 테넌트 이름을 가져오는 방법을 보여줘요.

public class GetTenantWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        String tenant = context.getTenant();
        System.out.println(tenant);

        return null;
    }

}

네임스페이스 가져오기 (Get namespace)

getNamespace 메서드는 윈도우 함수와 연결된 네임스페이스를 가져와요. 이 예시는 Java 윈도우 함수에서 네임스페이스를 가져오는 방법을 보여줘요.

public class GetNamespaceWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        String ns = context.getNamespace();
        System.out.println(ns);

        return null;
    }

}

함수 이름 가져오기 (Get function name)

getFunctionName 메서드는 윈도우 함수 이름을 가져와요. 이 예시는 Java 윈도우 함수에서 함수 이름을 가져오는 방법을 보여줘요.

public class GetNameOfWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        String functionName = context.getFunctionName();
        System.out.println(functionName);

        return null;
    }

}

함수 ID 가져오기 (Get function ID)

getFunctionId 메서드는 윈도우 함수 ID를 가져와요. 이 예시는 Java 윈도우 함수에서 함수 ID를 가져오는 방법을 보여줘요.

public class GetFunctionIDWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        String functionID = context.getFunctionId();
        System.out.println(functionID);

        return null;
    }

}

함수 버전 가져오기 (Get function version)

getFunctionVersion 메서드는 윈도우 함수 버전을 가져와요. 이 예시는 Java 윈도우 함수의 함수 버전을 가져오는 방법을 보여줘요.

public class GetVersionOfWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        String functionVersion = context.getFunctionVersion();
        System.out.println(functionVersion);

        return null;
    }

}

인스턴스 ID 가져오기 (Get instance ID)

getInstanceId 메서드는 윈도우 함수의 인스턴스 ID를 가져와요. 이 예시는 Java 윈도우 함수에서 인스턴스 ID를 가져오는 방법을 보여줘요.

public class GetInstanceIDWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        int instanceId = context.getInstanceId();
        System.out.println(instanceId);

        return null;
    }

}

인스턴스 수 가져오기 (Get num instances)

getNumInstances 메서드는 윈도우 함수를 호출하는 인스턴스의 수를 가져와요. 이 예시는 Java 윈도우 함수에서 인스턴스 수를 가져오는 방법을 보여줘요.

public class GetNumInstancesWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        int numInstances = context.getNumInstances();
        System.out.println(numInstances);

        return null;
    }

}

출력 스키마 타입 가져오기 (Get output schema type)

getOutputSchemaType 메서드는 출력 스키마의 내장형 타입 또는 사용자 정의 클래스 이름을 가져와요. 이 예시는 Java 윈도우 함수의 출력 스키마 타입을 가져오는 방법을 보여줘요.

public class GetOutputSchemaTypeWindowFunction implements WindowFunction<String, Void> {

    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        String schemaType = context.getOutputSchemaType();
        System.out.println(schemaType);

        return null;
    }
}

Logger

Java SDK를 사용하는 Pulsar 윈도우 함수는 선택한 로그 레벨로 로그를 생산하는 데 쓸 수 있는 SLF4j Logger 객체에 접근할 수 있어요. 이 예시는 들어오는 문자열에 danger 단어가 있는지 여부에 따라 Java 함수에서 WARNING 레벨 또는 INFO 레벨 로그를 기록해요.

import java.util.Collection;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.api.WindowContext;
import org.apache.pulsar.functions.api.WindowFunction;
import org.slf4j.Logger;

public class LoggingWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        Logger log = context.getLogger();
        for (Record<String> record : inputs) {
            log.info(record + "-window-log");
        }
        return null;
    }

}

함수가 로그를 생산해야 한다면, 함수를 만들거나 실행할 때 로그 토픽을 지정하세요.

bin/pulsar-admin functions create \
  --jar $PWD/my-functions.jar \
  --classname my.package.LoggingFunction \
  --log-topic persistent://public/default/logging-function-logs \
  # Other function configs

LoggingFunction이 생산한 모든 로그는 persistent://public/default/logging-function-logs 토픽으로 접근할 수 있어요.

Metrics

Pulsar 윈도우 함수는 조회 가능한 메트릭 인터페이스에 임의의 메트릭을 게시할 수 있어요. Java의 언어 네이티브 인터페이스를 사용하는 Pulsar 윈도우 함수는 Pulsar에 메트릭과 통계를 게시할 수 없어요. 컨텍스트 객체를 사용해 키별로 메트릭을 기록할 수 있어요. 이 예시는 Java 함수에서 함수가 메시지를 처리할 때마다 process-count 키용 메트릭과 elevens-count 키용 메트릭을 따로 설정해요.

import java.util.Collection;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.api.WindowContext;
import org.apache.pulsar.functions.api.WindowFunction;

/**
 * Example function that wants to keep track of
 * the event time of each message sent.
 */
public class UserMetricWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {

        for (Record<String> record : inputs) {
            if (record.getEventTime().isPresent()) {
                context.recordMetric("MessageEventTime", record.getEventTime().get().doubleValue());
            }
        }

        return null;
    }
}

사용자 설정 (User config)

SDK로 만든 Pulsar Functions를 실행하거나 업데이트할 때, --user-config 플래그로 임의의 키/값 쌍을 전달할 수 있어요. 키/값 쌍은 JSON으로 지정해야 해요. 이 예시는 함수에 사용자 구성 키/값을 전달해요.

bin/pulsar-admin functions create \
  --name word-filter \
 --user-config '{"forbidden-word":"rosebud"}' \
  # Other function configs

API

윈도우 함수의 사용자 정의 정보를 가져오려면 다음 API를 사용할 수 있어요.

getUserConfigMap

getUserConfigMap API는 윈도우 함수의 모든 사용자 정의 키/값 설정 맵을 가져와요.

/**
     * Get a map of all user-defined key/value configs for the function.
     *
     * @return The full map of user-defined config values
     */
    Map<String, Object> getUserConfigMap();

getUserConfigValue

getUserConfigValue API는 사용자 정의 키/값을 가져와요.

/**
     * Get any user-defined key/value.
     *
     * @param key The key
     * @return The Optional value specified by the user for that key.
     */
    Optional<Object> getUserConfigValue(String key);

getUserConfigValueOrDefault

getUserConfigValueOrDefault API는 사용자 정의 키/값을 가져오거나, 없으면 기본값을 가져와요.

/**
     * Get any user-defined key/value or a default value if none is present.
     *
     * @param key
     * @param defaultValue
     * @return Either the user config value associated with a given key or a supplied default value
     */
    Object getUserConfigValueOrDefault(String key, Object defaultValue);

이 예시는 Pulsar 윈도우 함수에 제공된 키/값 쌍에 접근하는 방법을 보여줘요. Java SDK 컨텍스트 객체는 명령줄(JSON)을 통해 Pulsar 윈도우 함수에 제공된 키/값 쌍에 접근할 수 있게 해줘요. Java 윈도우 함수에 전달되는 모든 키/값 쌍에서 키와 값 모두 String이에요. 값을 다른 타입으로 설정하려면 String 타입에서 역직렬화해야 해요. 이 예시는 Java 윈도우 함수에서 키/값 쌍을 전달해요.

bin/pulsar-admin functions create \
   --user-config '{"word-of-the-day":"verdure"}' \
  # Other function configs

이 예시는 Java 윈도우 함수에서 값에 접근해요. UserConfigFunction 함수는 호출될 때마다(즉 메시지가 도착할 때마다) "The word of the day is verdure" 문자열을 로그로 남겨요. word-of-the-day의 사용자 설정은 명령줄 도구나 REST API 등 여러 방법을 통해 함수가 새 설정 값으로 업데이트될 때만 변경돼요.

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

import java.util.Optional;

public class UserConfigWindowFunction implements WindowFunction<String, String> {
    @Override
    public String process(Collection<Record<String>> input, WindowContext context) throws Exception {
        Optional<Object> whatToWrite = context.getUserConfigValue("WhatToWrite");
        if (whatToWrite.get() != null) {
            return (String)whatToWrite.get();
        } else {
            return "Not a nice way";
        }
    }
}

값이 제공되지 않으면 전체 사용자 설정 맵에 접근하거나 기본값을 설정할 수 있어요.

// Get the whole config map
Map<String, String> allConfigs = context.getUserConfigMap();

// Get value or resort to default
String wotd = context.getUserConfigValueOrDefault("word-of-the-day", "perspicacious");

라우팅 (Routing)

context.publish() 인터페이스로 원하는 만큼 많은 결과를 게시할 수 있어요. 이 예시는 Java 함수에서 PublishFunction 클래스가 컨텍스트의 내장 함수를 사용해 publishTopic에 메시지를 게시하는 것을 보여줘요.

public class PublishWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> input, WindowContext context) throws Exception {
        String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic");
        String output = String.format("%s!", input);
        context.publish(publishTopic, output);

        return null;
    }

}

상태 저장 (State storage)

Pulsar 윈도우 함수는 상태 저장 인터페이스로 Apache BookKeeper를 사용해요. Apache Pulsar 설치(단일 설치 포함)에는 BookKeeper bookie 배포가 포함돼 있어요.

Apache Pulsar는 함수의 상태를 저장하기 위해 Apache BookKeeper 테이블 서비스와 통합돼요. 예를 들어 WordCount 함수는 Pulsar Functions 상태 API를 통해 자신의 카운터 상태를 BookKeeper 테이블 서비스에 저장할 수 있어요.

상태는 키/값 쌍이며, 키는 문자열이고 값은 임의의 바이너리 데이터예요. 카운터는 64비트 빅엔디언 바이너리 값으로 저장돼요. 키는 개별 Pulsar Function으로 범위가 지정되고 해당 함수의 인스턴스들 사이에서 공유돼요.

현재 Pulsar 윈도우 함수는 상태에 접근·업데이트·관리하는 Java API를 노출해요. Java SDK 함수를 사용할 때 이 API는 컨텍스트 객체에서 사용할 수 있어요.

Java API	설명
incrCounter	키가 참조하는 내장 분산 카운터를 증가시켜요.
getCounter	키의 카운터 값을 가져와요.
putState	키의 상태 값을 업데이트해요.

Java 윈도우 함수에서 상태에 접근·업데이트·관리하기 위해 다음 API를 사용할 수 있어요.

incrCounter

incrCounter API는 키가 참조하는 내장 분산 카운터를 증가시켜요. 애플리케이션은 incrCounter API를 사용해 주어진 키의 카운터를 주어진 만큼 변경해요. 키가 존재하지 않으면 새 키가 만들어져요.

    /**
     * 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);

getCounter

getCounter API는 키의 카운터 값을 가져와요. 애플리케이션은 getCounter API를 사용해 incrCounter API로 변경된 주어진 키의 카운터를 검색해요.

    /**
     * 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);

getCounter API 외에도, Pulsar는 함수가 일반 키/값 상태를 저장할 수 있도록 일반 키/값 API(putState)도 노출해요.

putState

putState API는 키의 상태 값을 업데이트해요.

    /**
     * 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);

이 예시는 애플리케이션이 Pulsar 윈도우 함수에서 상태를 저장하는 방법을 보여줘요. WordCountWindowFunction의 로직은 단순하고 직관적이에요.

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

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

import java.util.Arrays;

public class WordCountWindowFunction implements WindowFunction<String, Void> {
    @Override
    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
        for (Record<String> input : inputs) {
            Arrays.asList(input.getValue().split("\\.")).forEach(word -> context.incrCounter(word, 1));
        }
        return null;

    }
}

더 알아보기 (Learn more)

• Pulsar Functions 개요 — Pulsar의 스트림 처리 단위인 Functions를 알아봐요.
• 함수 컨텍스트 — 일반 함수 컨텍스트에서 제공하는 정보를 살펴봐요.
• 윈도우 함수 — 윈도우 함수가 무엇인지, 어떻게 동작하는지 알아봐요.
• 상태 저장 — 함수 상태 저장 개념을 자세히 살펴봐요.
• Pulsar Functions 개발 — Java SDK로 함수를 개발하는 방법을 알아봐요.