This hands-on tutorial provides step-by-step instructions and examples on how to create and validate functions in a standalone Pulsar, including stateful functions and window functions.
Enable pulsar function in conf/standalone.conf (add this field if not exists):
functionsWorkerEnabled=true
Start Pulsar locally.
bin/pulsar standalone
All the components (including ZooKeeper, BookKeeper, broker, and so on) of a Pulsar service start in order. You can use the bin/pulsar-admin brokers healthcheck command to make sure the Pulsar service is up and running.
You can see both the example-function-config.yaml and api-examples.jar files under the examples folder of the Pulsar’s directory on your local machine.
This example function will add a ! at the end of every message.
You can check the config of this function in examples/example-function-config.yaml:
tenant:"test" namespace:"test-namespace" name:"example"# function name className:"org.apache.pulsar.functions.api.examples.ExclamationFunction" inputs:["test_src"]# this function will read messages from these topics output:"test_result"# the return value of this function will be sent to this topic autoAck:true# function will acknowledge input messages if set true parallelism:1
You can see the source code of ExclamationFunction.
For more information about the yaml config, see the reference.
In the same terminal window as step 1, verify the function's configurations.
bin/pulsar-admin functions get \ --tenanttest\ --namespace test-namespace \ --name example
In a new terminal window, produce messages to the input topictest_src.
bin/pulsar-client produce -m"test-messages-`date`"-n10 test_src
In the same terminal window as step 1, the messages produced by the example function are returned. You can see there is a ! added at the end of every message.
You can check the config of this function in examples/example-stateful-function-config.yaml:
tenant:"test" namespace:"test-namespace" name:"word_count" className:"org.apache.pulsar.functions.api.examples.WordCountFunction" inputs:["test_wordcount_src"]# this function will read messages from these topics autoAck:true parallelism:1
You can see the source code of WordCountFunction. This function won't return any value but store the occurrence of words in function context. So you don't need to specify an output topic.
For more information about the yaml config, see the reference.
In the same terminal window as step 1, get the information of the word_count function.
bin/pulsar-admin functions get \ --tenanttest\ --namespace test-namespace \ --name word_count
In the same terminal window as step 1, query the state table for the function with the key hello. This operation watches the changes associated with hello.
For more information about the pulsar-admin functions querystate options command, including flags, descriptions, default values, and shorthands, see Pulsar admin docs.
In a new terminal window, produce 10 messages with hello to the input topictest_wordcount_src using one of the following methods. The value of hello is updated to 10.
bin/pulsar-client produce -m"hello"-n10 test_wordcount_src
In the same terminal window as step 1, check the result.
The result shows that the output topictest_wordcount_dest receives the messages.
Output
{ "key":"hello", "numberValue":10, "version":9 }
In the terminal window as step 5, produce another 10 messages with hello. The value of hello is updated to 20.
bin/pulsar-client produce -m"hello"-n10 test_wordcount_src
In the same terminal window as step 1, check the result.
The result shows that the output topictest_wordcount_dest receives the value of 20.