aggregate-functions - 跨 Apache Pulsar 函数访问状态

Question

我希望有一个函数聚合事件并在簿记员中维护状态，同时允许处理不同事件的其他函数利用该状态（通过键访问，或通过键来查找状态。

我无法通过上下文对象找到任何方法，有不同的方法吗？

Answer 1

您可以利用

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

和

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

只要您协调将在函数之间使用的键，上下文对象的方法即可完成此操作。聚合函数将使用 putState 方法和预先确定的键执行计算并存储数据，如下所示：

public class AggregateFunction implements Function<String, Void> {
@Override
public Void process(String input, Context context) {
    ByteBuffer value;
    // Calculate value and place in ByteBuffer
    context.putState("PRE-DETERMINED-KEY", value);
}

}

然后消费函数可以像这样访问这个值；

public class ConsumingFunction implements Function<String, String> {
@Override
public String process(String input, Context context) {
    ByteBuffer value = context.gettState("PRE-DETERMINED-KEY");
    // Perform logic based on the value.
    return "";
}

}