Getting started with Pulsar Functions

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.

Prerequisites

Start standalone Pulsar

  1. Enable pulsar function in conf/standalone.conf (add this field if not exists):

    1. functionsWorkerEnabled=true
  2. Start Pulsar locally.

    1. 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.

  3. Check the Pulsar binary protocol port.

    1. telnet localhost 6650
  4. Check the Pulsar Function cluster.

    1. bin/pulsar-admin functions-worker get-cluster

    Output

    1. [{"workerId":"c-standalone-fw-localhost-6750","workerHostname":"localhost","port":6750}]
  5. Make sure a public tenant exists.

    1. bin/pulsar-admin tenants list

    Output

    1. public
  6. Make sure a default namespace exists.

    1. bin/pulsar-admin namespaces list public

    Output

    1. public/default
  7. Make sure the table service is enabled successfully.

    1. telnet localhost 4181

    Output

    1. Trying ::1...
    2. telnet: connect to address ::1: Connection refused
    3. Trying 127.0.0.1...
    4. Connected to localhost.
    5. Escape character is '^]'.

Create a namespace for test

  1. Create a tenant and a namespace.

    1. bin/pulsar-admin tenants create test
    2. bin/pulsar-admin namespaces create test/test-namespace
  2. In the same terminal window as step 1, verify the tenant and the namespace.

    1. bin/pulsar-admin namespaces list test

    Output

    This output shows that both tenant and namespace are created successfully.

    1. "test/test-namespace"

Start functions

Get started - 图1note

Before starting functions, you need to start Pulsar and create a test namespace.

  1. Create a function named examples.

    Get started - 图2tip

    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.

    1. bin/pulsar-admin functions create \
    2. --function-config-file examples/example-function-config.yaml \
    3. --jar examples/api-examples.jar

    Output

    1. Created Successfully

    You can check the config of this function in examples/example-function-config.yaml:

    1. tenant: "test"
    2. namespace: "test-namespace"
    3. name: "example" # function name
    4. className: "org.apache.pulsar.functions.api.examples.ExclamationFunction"
    5. inputs: ["test_src"] # this function will read messages from these topics
    6. output: "test_result" # the return value of this function will be sent to this topic
    7. autoAck: true # function will acknowledge input messages if set true
    8. parallelism: 1

    You can see the source code of ExclamationFunction. For more information about the yaml config, see the reference.

  2. In the same terminal window as step 1, verify the function’s configurations.

    1. bin/pulsar-admin functions get \
    2. --tenant test \
    3. --namespace test-namespace \
    4. --name example

    Output

    1. {
    2. "tenant": "test",
    3. "namespace": "test-namespace",
    4. "name": "example",
    5. "className": "org.apache.pulsar.functions.api.examples.ExclamationFunction",
    6. "inputSpecs": {
    7. "test_src": {
    8. "isRegexPattern": false,
    9. "schemaProperties": {},
    10. "consumerProperties": {},
    11. "poolMessages": false
    12. }
    13. },
    14. "output": "test_result",
    15. "producerConfig": {
    16. "useThreadLocalProducers": false,
    17. "batchBuilder": ""
    18. },
    19. "processingGuarantees": "ATLEAST_ONCE",
    20. "retainOrdering": false,
    21. "retainKeyOrdering": false,
    22. "forwardSourceMessageProperty": true,
    23. "userConfig": {},
    24. "runtime": "JAVA",
    25. "autoAck": true,
    26. "parallelism": 1,
    27. "resources": {
    28. "cpu": 1.0,
    29. "ram": 1073741824,
    30. "disk": 10737418240
    31. },
    32. "cleanupSubscription": true,
    33. "subscriptionPosition": "Latest"
    34. }
  3. In the same terminal window as step 1, verify the function’s status.

    1. bin/pulsar-admin functions status \
    2. --tenant test \
    3. --namespace test-namespace \
    4. --name example

    Output

    "running": true shows that the function is running.

    1. {
    2. "numInstances" : 1,
    3. "numRunning" : 1,
    4. "instances" : [ {
    5. "instanceId" : 0,
    6. "status" : {
    7. "running" : true,
    8. "error" : "",
    9. "numRestarts" : 0,
    10. "numReceived" : 0,
    11. "numSuccessfullyProcessed" : 0,
    12. "numUserExceptions" : 0,
    13. "latestUserExceptions" : [ ],
    14. "numSystemExceptions" : 0,
    15. "latestSystemExceptions" : [ ],
    16. "averageLatency" : 0.0,
    17. "lastInvocationTime" : 0,
    18. "workerId" : "c-standalone-fw-localhost-8080"
    19. }
    20. } ]
    21. }
  4. In the same terminal window as step 1, subscribe to the output topic test_result.

    1. bin/pulsar-client consume -s test-sub -n 0 test_result
  5. In a new terminal window, produce messages to the input topic test_src.

    1. bin/pulsar-client produce -m "test-messages-`date`" -n 10 test_src
  6. 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.

    Output

    1. ----- got message -----
    2. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    3. ----- got message -----
    4. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    5. ----- got message -----
    6. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    7. ----- got message -----
    8. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    9. ----- got message -----
    10. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    11. ----- got message -----
    12. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    13. ----- got message -----
    14. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    15. ----- got message -----
    16. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    17. ----- got message -----
    18. test-messages-Thu Jul 19 11:59:15 PDT 2021!
    19. ----- got message -----
    20. test-messages-Thu Jul 19 11:59:15 PDT 2021!

Start stateful functions

The standalone mode of Pulsar enables BookKeeper table service for stateful functions. For more information, see Configure state storage.

The following example provides instructions to validate counter functions.

Get started - 图3note

Before starting stateful functions, you need to start Pulsar and create a test namespace.

  1. Create a function using examples/example-stateful-function-config.yaml.

    1. bin/pulsar-admin functions create \
    2. --function-config-file examples/example-stateful-function-config.yaml \
    3. --jar examples/api-examples.jar

    Output

    1. Created Successfully

    You can check the config of this function in examples/example-stateful-function-config.yaml:

    1. tenant: "test"
    2. namespace: "test-namespace"
    3. name: "word_count"
    4. className: "org.apache.pulsar.functions.api.examples.WordCountFunction"
    5. inputs: ["test_wordcount_src"] # this function will read messages from these topics
    6. autoAck: true
    7. 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.

  2. In the same terminal window as step 1, get the information of the word_count function.

    1. bin/pulsar-admin functions get \
    2. --tenant test \
    3. --namespace test-namespace \
    4. --name word_count

    Output

    1. {
    2. "tenant": "test",
    3. "namespace": "test-namespace",
    4. "name": "word_count",
    5. "className": "org.apache.pulsar.functions.api.examples.WordCountFunction",
    6. "inputSpecs": {
    7. "test_wordcount_src": {
    8. "isRegexPattern": false,
    9. "schemaProperties": {},
    10. "consumerProperties": {},
    11. "poolMessages": false
    12. }
    13. },
    14. "producerConfig": {
    15. "useThreadLocalProducers": false,
    16. "batchBuilder": ""
    17. },
    18. "processingGuarantees": "ATLEAST_ONCE",
    19. "retainOrdering": false,
    20. "retainKeyOrdering": false,
    21. "forwardSourceMessageProperty": true,
    22. "userConfig": {},
    23. "runtime": "JAVA",
    24. "autoAck": true,
    25. "parallelism": 1,
    26. "resources": {
    27. "cpu": 1.0,
    28. "ram": 1073741824,
    29. "disk": 10737418240
    30. },
    31. "cleanupSubscription": true,
    32. "subscriptionPosition": "Latest"
    33. }
  3. In the same terminal window as step 1, get the status of the word_count function.

    1. bin/pulsar-admin functions status \
    2. --tenant test \
    3. --namespace test-namespace \
    4. --name word_count

    Output

    1. {
    2. "numInstances" : 1,
    3. "numRunning" : 1,
    4. "instances" : [ {
    5. "instanceId" : 0,
    6. "status" : {
    7. "running" : true,
    8. "error" : "",
    9. "numRestarts" : 0,
    10. "numReceived" : 0,
    11. "numSuccessfullyProcessed" : 0,
    12. "numUserExceptions" : 0,
    13. "latestUserExceptions" : [ ],
    14. "numSystemExceptions" : 0,
    15. "latestSystemExceptions" : [ ],
    16. "averageLatency" : 0.0,
    17. "lastInvocationTime" : 0,
    18. "workerId" : "c-standalone-fw-localhost-8080"
    19. }
    20. } ]
    21. }
  4. 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.

    1. bin/pulsar-admin functions querystate \
    2. --tenant test \
    3. --namespace test-namespace \
    4. --name word_count -k hello -w

    Get started - 图4tip

    For more information about the pulsar-admin functions querystate options command, including flags, descriptions, default values, and shorthands, see Pulsar admin docs.

    Output

    1. key 'hello' doesn't exist.
    2. key 'hello' doesn't exist.
    3. key 'hello' doesn't exist.
    4. ...
  5. In a new terminal window, produce 10 messages with hello to the input topic test_wordcount_src using one of the following methods. The value of hello is updated to 10.

    1. bin/pulsar-client produce -m "hello" -n 10 test_wordcount_src
  6. In the same terminal window as step 1, check the result.

    The result shows that the output topic test_wordcount_dest receives the messages.

    Output

    1. {
    2. "key": "hello",
    3. "numberValue": 10,
    4. "version": 9
    5. }
  7. In the terminal window as step 5, produce another 10 messages with hello. The value of hello is updated to 20.

    1. bin/pulsar-client produce -m "hello" -n 10 test_wordcount_src
  8. In the same terminal window as step 1, check the result.

    The result shows that the output topic test_wordcount_dest receives the value of 20.

    1. {
    2. "key": "hello",
    3. "numberValue": 20,
    4. "version": 19
    5. }

Start window functions

Window functions are a special form of Pulsar Functions. For more information, see concepts.

The following example provides instructions to start a window function to calculate the sum in a window.

Get started - 图5note

Before starting window functions, you need to start Pulsar and create a test namespace.

  1. Create a function using example-window-function-config.yaml.

    1. bin/pulsar-admin functions create \
    2. --function-config-file examples/example-window-function-config.yaml \
    3. --jar examples/api-examples.jar

    Output

    1. Created Successfully

    You can check the config of this function in examples/example-window-function-config.yaml:

    1. tenant: "test"
    2. namespace: "test-namespace"
    3. name: "window-example"
    4. className: "org.apache.pulsar.functions.api.examples.AddWindowFunction"
    5. inputs: ["test_window_src"]
    6. output: "test_window_result"
    7. autoAck: true
    8. parallelism: 1
    9. # every 5 messages, calculate sum of the latest 10 messages
    10. windowConfig:
    11. windowLengthCount: 10
    12. slidingIntervalCount: 5

    You can see the source code of AddWindowFunction here. For more information about the yaml config, see the reference.

  2. In the same terminal window as step 1, verify the function’s configurations.

    1. bin/pulsar-admin functions get \
    2. --tenant test \
    3. --namespace test-namespace \
    4. --name window-example

    Output

    1. {
    2. "tenant": "test",
    3. "namespace": "test-namespace",
    4. "name": "window-example",
    5. "className": "org.apache.pulsar.functions.api.examples.AddWindowFunction",
    6. "inputSpecs": {
    7. "test_window_src": {
    8. "isRegexPattern": false,
    9. "schemaProperties": {},
    10. "consumerProperties": {},
    11. "poolMessages": false
    12. }
    13. },
    14. "output": "test_window_result",
    15. "producerConfig": {
    16. "useThreadLocalProducers": false,
    17. "batchBuilder": ""
    18. },
    19. "processingGuarantees": "ATLEAST_ONCE",
    20. "retainOrdering": false,
    21. "retainKeyOrdering": false,
    22. "forwardSourceMessageProperty": true,
    23. "userConfig": {},
    24. "runtime": "JAVA",
    25. "autoAck": false,
    26. "parallelism": 1,
    27. "resources": {
    28. "cpu": 1.0,
    29. "ram": 1073741824,
    30. "disk": 10737418240
    31. },
    32. "windowConfig": {
    33. "windowLengthCount": 10,
    34. "slidingIntervalCount": 5,
    35. "actualWindowFunctionClassName": "org.apache.pulsar.functions.api.examples.AddWindowFunction",
    36. "processingGuarantees": "ATLEAST_ONCE"
    37. },
    38. "cleanupSubscription": true,
    39. "subscriptionPosition": "Latest"
    40. }
  3. In the same terminal window as step 1, verify the function’s status.

    1. bin/pulsar-admin functions status \
    2. --tenant test \
    3. --namespace test-namespace \
    4. --name window-example

    Output

    "running": true shows that the function is running.

    1. {
    2. "numInstances" : 1,
    3. "numRunning" : 1,
    4. "instances" : [ {
    5. "instanceId" : 0,
    6. "status" : {
    7. "running" : true,
    8. "error" : "",
    9. "numRestarts" : 0,
    10. "numReceived" : 0,
    11. "numSuccessfullyProcessed" : 0,
    12. "numUserExceptions" : 0,
    13. "latestUserExceptions" : [ ],
    14. "numSystemExceptions" : 0,
    15. "latestSystemExceptions" : [ ],
    16. "averageLatency" : 0.0,
    17. "lastInvocationTime" : 0,
    18. "workerId" : "c-standalone-fw-localhost-8080"
    19. }
    20. } ]
    21. }
  4. In the same terminal window as step 1, subscribe to the output topic test_window_result.

    1. bin/pulsar-client consume -s test-sub -n 0 test_window_result
  5. In a new terminal window, produce messages to the input topic test_window_src.

    1. bin/pulsar-client produce -m "3" -n 10 test_window_src
  6. In the same terminal window as step 1, the messages produced by the window function window-example are returned.

    Output

    1. ----- got message -----
    2. key:[null], properties:[], content:15
    3. ----- got message -----
    4. key:[null], properties:[], content:30