Get started with Pulsar Functions

This tutorial walks you through running a standalone Pulsar cluster on your machine, and then running your first Pulsar Function using that cluster. The first Pulsar Function runs in local run mode (outside your Pulsar cluster), while the second runs in cluster mode (inside your cluster).

In local run mode, Pulsar Functions communicate with Pulsar cluster, but run outside of the cluster.

Prerequisites

Install Maven on your machine.

Run a standalone Pulsar cluster

In order to run Pulsar Functions, you need to run a Pulsar cluster locally first. The easiest way is to run Pulsar in standalone mode. Follow these steps to start up a standalone cluster.

  1. $ wget https://archive.apache.org/dist/pulsar/pulsar-2.4.0/apache-pulsar-2.4.0-bin.tar.gz
  2. $ tar xvfz apache-pulsar-2.4.0-bin.tar.gz
  3. $ cd apache-pulsar-2.4.0
  4. $ bin/pulsar standalone \
  5.   --advertised-address 127.0.0.1

When running Pulsar in standalone mode, the public tenant and the default namespace are created automatically. The tenant and namespace are used throughout this tutorial.

Run a Pulsar Function in local run mode

You can start with a simple function that takes a string as input from a Pulsar topic, adds an exclamation point to the end of the string, and then publishes the new string to another Pulsar topic. The following is the code for the function.

  1. package org.apache.pulsar.functions.api.examples;
  3. import java.util.function.Function;
  5. public class ExclamationFunction implements Function<String, String> {
  6.     @Override
  7.     public String apply(String input) {
  8.         return String.format("%s!", input);
  9.     }
  10. }

A JAR file containing this function and several other functions (written in Java) is included with the binary distribution you have downloaded (in the examples folder). Run the function in local mode on your laptop but outside your Pulsar cluster with the following commands.

  1. $ bin/pulsar-admin functions localrun \
  2.   --jar examples/api-examples.jar \
  3.   --classname org.apache.pulsar.functions.api.examples.ExclamationFunction \
  4.   --inputs persistent://public/default/exclamation-input \
  5.   --output persistent://public/default/exclamation-output \
  6.   --name exclamation

Multiple input topics

In the example above, a single topic is specified using the —inputs flag. You can also specify multiple input topics with a comma-separated list using the same flag.

  1. —inputs topic1,topic2

You can open up another shell and use the pulsar-client tool to listen for messages on the output topic.

  1. $ bin/pulsar-client consume persistent://public/default/exclamation-output \
  2.   --subscription-name my-subscription \
  3.   --num-messages 0

Setting the —num-messages flag to 0 means that consumers listen on the topic indefinitely, rather than only accepting a certain number of messages.

With a listener up and running, you can open up another shell and produce a message on the input topic that you specify.

  1. $ bin/pulsar-client produce persistent://public/default/exclamation-input \
  2.   --num-produce 1 \
  3.   --messages "Hello world"

When the message has been successfully processed by the exclamation function, you will see the following output. To shut down the function, press Ctrl+C.

  1. ----- got message -----
  2. Hello world!

Process explanation

  • The Hello world message you publish to the input topic (persistent://public/default/exclamation-input) is passed to the exclamation function.
  • The exclamation function processes the message (providing a result of Hello world!) and publishes the result to the output topic (persistent://public/default/exclamation-output).
  • If the exclamation function does not run, Pulsar will durably store the message data published to the input topic in Apache BookKeeper until a consumer consumes and acknowledges the message.

Run a Pulsar Function in cluster mode

Local run mode is useful for development and test. However, when you use Pulsar for real deployment, you run it in cluster mode. In cluster mode, Pulsar Functions run inside of your Pulsar cluster and are managed using the same pulsar-admin functions interface.

The following command deploys the same exclamation function you run locally in your Pulsar cluster, rather than outside of it.

  1. $ bin/pulsar-admin functions create \
  2.   --jar examples/api-examples.jar \
  3.   --classname org.apache.pulsar.functions.api.examples.ExclamationFunction \
  4.   --inputs persistent://public/default/exclamation-input \
  5.   --output persistent://public/default/exclamation-output \
  6.   --name exclamation

You will see Created successfully in the output. Check the list of functions running in your cluster.

  1. $ bin/pulsar-admin functions list \
  2.   --tenant public \
  3.   --namespace default

You will see the exclamation function. Check the status of your deployed function using the getstatus command.

  1. $ bin/pulsar-admin functions getstatus \
  2.   --tenant public \
  3.   --namespace default \
  4.   --name exclamation

You will see the following JSON output.

  1. {
  2.   "functionStatusList": [
  3.     {
  4.       "running": true,
  5.       "instanceId": "0"
  6.     }
  7.   ]
  8. }

As you can see, the instance is currently running, and an instance with the ID of 0 is running. With the get command, you can get other information about the function, for example, topics, tenant, namespace, and so on.

  1. $ bin/pulsar-admin functions get \
  2.   --tenant public \
  3.   --namespace default \
  4.   --name exclamation

You will see the following JSON output.

  1. {
  2.   "tenant": "public",
  3.   "namespace": "default",
  4.   "name": "exclamation",
  5.   "className": "org.apache.pulsar.functions.api.examples.ExclamationFunction",
  6.   "output": "persistent://public/default/exclamation-output",
  7.   "autoAck": true,
  8.   "inputs": [
  9.     "persistent://public/default/exclamation-input"
  10.   ],
  11.   "parallelism": 1
  12. }

As you can see, only one instance of the function is running in your cluster. Update the parallel functions to 3 using the update command.

  1. $ bin/pulsar-admin functions update \
  2.   --jar examples/api-examples.jar \
  3.   --classname org.apache.pulsar.functions.api.examples.ExclamationFunction \
  4.   --inputs persistent://public/default/exclamation-input \
  5.   --output persistent://public/default/exclamation-output \
  6.   --tenant public \
  7.   --namespace default \
  8.   --name exclamation \
  9.   --parallelism 3

You will see Updated successfully in the output. If you enter the get command, you see that the parallel functions are increased to 3, meaning that three instances of the function are running in your cluster.

  1. {
  2.   "tenant": "public",
  3.   "namespace": "default",
  4.   "name": "exclamation",
  5.   "className": "org.apache.pulsar.functions.api.examples.ExclamationFunction",
  6.   "output": "persistent://public/default/exclamation-output",
  7.   "autoAck": true,
  8.   "inputs": [
  9.     "persistent://public/default/exclamation-input"
  10.   ],
  11.   "parallelism": 3
  12. }

Shut down the running function with the delete command.

  1. $ bin/pulsar-admin functions delete \
  2.   --tenant public \
  3.   --namespace default \
  4.   --name exclamation

When you see Deleted successfully in the output, you've successfully run, updated, and shut down functions running in cluster mode.

Write and run a new function

In order to write and run Python functions, you need to install some dependencies.

  1. $ pip install pulsar-client

In the examples above, you run and manage pre-written Pulsar Functions and learn how they work. You can also write your own functions with Python API. In the following example, the function takes a string as input, reverses the string, and publishes the reversed string to the specified topic.

First, create a new Python file.

  1. $ touch reverse.py

Add the following information in the Python file.

  1. def process(input):
  2.     return input[::-1]

The process method defines the processing logic of Pulsar Functions. It uses Python slice magic to reverse each incoming string. You can deploy the function using the create command.

  1. $ bin/pulsar-admin functions create \
  2.   --py reverse.py \
  3.   --classname reverse \
  4.   --inputs persistent://public/default/backwards \
  5.   --output persistent://public/default/forwards \
  6.   --tenant public \
  7.   --namespace default \
  8.   --name reverse

If you see Created successfully, the function is ready to accept incoming messages. Because the function is running in cluster mode, you can trigger the function using the trigger command. This command sends a message that you specify to the function and returns the function output. The following is an example.

  1. $ bin/pulsar-admin functions trigger \
  2.   --name reverse \
  3.   --tenant public \
  4.   --namespace default \
  5.   --trigger-value "sdrawrof won si tub sdrawkcab saw gnirts sihT"

You will get the following output.

  1. This string was backwards but is now forwards

You have created a new Pulsar Function, deployed it in your Pulsar standalone cluster in cluster mode, and triggered the Function.

Write and run a Go function

Go function depends on pulsar-client-go. Make sure that you have built pulsar-client-go before using Go function.

To write and run a Go function, complete the following steps.

  • Create a new Go file.
  1. touch helloFunc.go
  • Append a byte for messages from the input topic.The following is a helloFunc.go example. Each message from the input topic is appended with a 110 byte, and then delivered to the output topic.
  1. package main
  3. import (
  4.     "context"
  6.     "github.com/apache/pulsar/pulsar-function-go/pf"
  7. )
  9. func HandleResponse(ctx context.Context, in []byte) ([]byte, error) {
  10.     res := append(in, 110)
  11.     return res, nil
  12. }
  14. func main() {
  15.     pf.Start(HandleResponse)
  16. }
  • Compile code.
  1. go build -o examplepulsar helloFunc.go
  • Run Go function.
  1. $ bin/pulsar-admin functions create \
  2.   --go examplepulsar \
  3.   --inputs persistent://public/default/backwards \
  4.   --output persistent://public/default/forwards \
  5.   --tenant public \
  6.   --namespace default \
  7.   --name gofunc

If you see Created successfully, the function is ready to accept incoming messages. Start a producer and produce messages to the backwards input topic. Start a consumer and consume messages from the forwards output topic, you will see 110 is appended to all messages.

The —classname parameter is not specified when running Go function, because there is no Class concept in Go, which is different from Java and Python.

NoteWhen you use the —go command to specify an executable file, make sure you have executable permissions.

Package Python dependencies

When you deploy Python functions in a cluster offline, you need to package the required dependencies in a ZIP file before deployment.

Client requirements

The following programs are required to be installed on the client machine.

  1. pip \\ required for getting python dependencies
  2. zip \\ for building zip archives

Python dependencies

A file named requirements.txt is needed with required dependencies for the Python function.

  1. sh==1.12.14

Prepare the Pulsar Function in the src folder.

Run the following command to gather Python dependencies in the deps folder.

  1. pip download \
  2. --only-binary :all: \
  3. --platform manylinux1_x86_64 \
  4. --python-version 27 \
  5. --implementation cp \
  6. --abi cp27m -r requirements.txt -d deps

Sample output

  1. Collecting sh==1.12.14 (from -r requirements.txt (line 1))
  2.   Using cached https://files.pythonhosted.org/packages/4a/22/17b22ef5b049f12080f5815c41bf94de3c229217609e469001a8f80c1b3d/sh-1.12.14-py2.py3-none-any.whl
  3.   Saved ./deps/sh-1.12.14-py2.py3-none-any.whl
  4. Successfully downloaded sh

Notepulsar-client is not needed as a dependency as it has already installed in the worker node.

Package

Create a destination folder with the desired package name, for example, exclamation. Copy the src and deps folders into it, and compress the folder into a ZIP archive.

Sample sequence

  1. cp -R deps exclamation/
  2. cp -R src exclamation/
  4. ls -la exclamation/
  5. total 7
  6. drwxr-xr-x   5 a.ahmed  staff  160 Nov  6 17:51 .
  7. drwxr-xr-x  12 a.ahmed  staff  384 Nov  6 17:52 ..
  8. drwxr-xr-x   3 a.ahmed  staff   96 Nov  6 17:51 deps
  9. drwxr-xr-x   3 a.ahmed  staff   96 Nov  6 17:51 src
  11. zip -r exclamation.zip exclamation

After package all the required dependencies into the exclamation.zip file, you can deploy functions in a Pulsar worker. The Pulsar worker does not need internet connectivity to download packages, because they are all included in the ZIP file.