Pulsar Functions

编程模型(Programming model)

开启Functions

  1. conf/bookkeeper.conf
extraServerComponents=org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent
  1. conf/broker.conf
functionsWorkerEnabled=true
  1. conf/functions_worker.yml
pulsarFunctionsCluster: pulsar-clusternumFunctionPackageReplicas: 2

窗口(window)

  • windowLengthCount 每个窗口的音讯数量
  • slidingIntervalCount 窗口滑动后的音讯数量
  • windowLengthDurationMs 窗口工夫
  • slidingIntervalDurationMs 窗口滑动后的工夫

开窗函数

public class WordCountWindowFunction implements org.apache.pulsar.functions.api.WindowFunction<String, Void> {    @Override    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {        for (Record<String> input : inputs) {        }        return null;    }}

运行函数

  • 工夫，滑动窗口

--user-config '{"windowLengthDurationMs":"60000", "slidingIntervalDurationMs":"1000"}'

  • 工夫，滚动窗口

--user-config '{"windowLengthDurationMs":"60000"}'

  • 数量，滑动窗口

--user-config '{"windowLengthCount":"100", "slidingIntervalCount":"10"}'

  • 数量，滚动窗口

--user-config '{"windowLengthCount":"100"}'

Java编程

pom.xml

<dependency>    <groupId>org.apache.pulsar</groupId>    <artifactId>pulsar-client</artifactId>    <version>${pulsar.version}</version></dependency><dependency>    <groupId>org.apache.pulsar</groupId>    <artifactId>pulsar-functions-api</artifactId>    <version>${pulsar.version}</version></dependency><dependency>    <groupId>org.apache.pulsar</groupId>    <artifactId>pulsar-functions-local-runner</artifactId>    <version>${pulsar.version}</version></dependency>
  1. WordCount
public class WordCountFunction implements org.apache.pulsar.functions.api.Function<String, Void> {    @Override    public Void process(String input, Context context) throws Exception {        Arrays.asList(input.split(" ")).forEach(word -> {            String counterKey = word.toLowerCase();            if (context.getCounter(counterKey) == 0) {                context.putState(counterKey, ByteBuffer.wrap(ByteUtils.from(100)));            }            context.incrCounter(counterKey, 1);        });        return null;    }}$ $PULSAR_HOME/bin/pulsar-admin functions create \--broker-service-url pulsar://server-101:6650 \--jar target/cloudwise-pulsar-functions-with-dependencies.jar \--classname com.cloudwise.quickstart.pulsar.functions.WordCountFunction \--tenant public \--namespace default \--name word-count-function \--inputs persistent://public/default/sentences \--output persistent://public/default/wordcount
  1. 动静路由
/** * 基本思路是查看每条音讯的内容，依据音讯内容将音讯路由到不同目的地。 */public class RoutingFunction implements org.apache.pulsar.functions.api.Function<String, String> {    @Override    public String process(String input, Context context) throws Exception {        String regex = context.getUserConfigValue("regex").toString();        String matchedTopic = context.getUserConfigValue("matched-topic").toString();        String unmatchedTopic = context.getUserConfigValue("unmatched-topic").toString();        Pattern pattern = Pattern.compile(regex);        Matcher matcher = pattern.matcher(input);        if (matcher.matches()) {            context.newOutputMessage(matchedTopic, Schema.STRING).value(input).send();        } else {            context.newOutputMessage(unmatchedTopic, Schema.STRING).value(input).send();        }        return null;    }}
  1. log-topic
public class LoggingFunction implements org.apache.pulsar.functions.api.Function<String, Void> {    @Override    public Void process(String s, Context context) throws Exception {        Logger LOG = context.getLogger();        String messageId = context.getFunctionId();        if (s.contains("danger")) {            LOG.warn("A warning was received in message {}", messageId);        } else {            LOG.info("Message {} received\nContent: {}", messageId, s);        }        return null;    }}$ $PULSAR_HOME/bin/pulsar-admin functions create \--jar cloudwise-pulsar-functions-1.0.0.jar \--classname com.cloudwise.quickstart.pulsar.functions.LoggingFunction \--log-topic persistent://public/default/logging-function-logs
  1. user-config
public class UserConfigFunction implements org.apache.pulsar.functions.api.Function<String, Void> {    @Override    public Void process(String s, Context context) throws Exception {        Logger log = context.getLogger();        Optional<Object> value = context.getUserConfigValue("word-of-the-day");        if (value.isPresent()) {            log.info("The word of the day is {}", value);        } else {            log.warn("No word of the day provided");        }        return null;    }}$ $PULSAR_HOME/bin/pulsar-admin functions create \--broker-service-url pulsar://server-101:6650 \--jar target/cloudwise-pulsar-functions-with-dependencies.jar \--classname com.cloudwise.quickstart.pulsar.functions.UserConfigFunction \--tenant public \--namespace default \--name word-count-function \--inputs persistent://public/default/userconfig \--user-config '{"word-of-the-day":"verdure"}'

更多福利

云智慧已开源集轻量级、聚合型、智能运维为一体的综合运维治理平台OMP（Operation Management Platform） ，具备 纳管、部署、监控、巡检、自愈、备份、复原 等性能，可为用户提供便捷的运维能力和业务管理，在进步运维人员等工作效率的同时，极大晋升了业务的连续性和安全性。点击下方地址链接，欢送大家给OMP点赞送star，理解更多相干内容～

GitHub地址：https://github.com/CloudWise-...

Gitee地址：https://gitee.com/CloudWise/OMP

微信扫描辨认下方二维码，备注【OMP】退出AIOps社区运维治理平台OMP开发者交换群，与更多行业大佬一起交流学习～

系列浏览

深入浅出Apache Pulsar（1）：Pulsar vs Kafka

深入浅出Apache Pulsar（2）：Pulsar音讯机制

深入浅出 Apache Pulsar（3）：Pulsar Schema