Use SerDe

Pulsar Functions use SerDe (Serialization and Deserialization) when publishing data to or consuming data from Pulsar topics. How SerDe works by default depends on the language you use (Java or Python) for a particular function. In both languages, however, you can write custom SerDe logic for more complex, application-specific types.

Use SerDe for Java functions

The following basic Java types are built-in and supported by default for Java functions: string, double, integer, float, long, short, and byte.

To customize Java types, you need to implement the following interface.

public interface SerDe<T> {
    T deserialize(byte[] input);
    byte[] serialize(T input);
}

SerDe works in the following ways for Java functions.

• If the input and output topics have a schema, Pulsar Functions use the schema for SerDe.
• If the input or output topics do not exist, Pulsar Functions adopt the following rules to determine SerDe:
  • If the schema type is specified, Pulsar Functions use the specified schema type.
  • If SerDe is specified, Pulsar Functions use the specified SerDe, and the schema type for input and output topics is byte.
  • If neither the schema type nor SerDe is specified, Pulsar Functions use the built-in SerDe. For non-primitive schema types, the built-in SerDe serializes and deserializes objects in the JSON format.

For example, imagine that you’re writing a function that processes tweet objects. You can refer to the following example of the Tweet class in Java.

public class Tweet {
    private String username;
    private String tweetContent;
    public Tweet(String username, String tweetContent) {
        this.username = username;
        this.tweetContent = tweetContent;
    }
    // Standard setters and getters
}

To pass Tweet objects directly between functions, you need to provide a custom SerDe class. In the example below, Tweet objects are basically strings, and username and tweet content are separated by |.

package com.example.serde;
import org.apache.pulsar.functions.api.SerDe;
import java.util.regex.Pattern;
public class TweetSerde implements SerDe<Tweet> {
    public Tweet deserialize(byte[] input) {
        String s = new String(input);
        String[] fields = s.split(Pattern.quote("|"));
        return new Tweet(fields[0], fields[1]);
    }
    public byte[] serialize(Tweet input) {
        return "%s|%s".format(input.getUsername(), input.getTweetContent()).getBytes();
    }
}

To apply a customized SerDe to a particular function, you need to:

• Package the Tweet and TweetSerde classes into a JAR.
• Specify a path to the JAR and SerDe class names when deploying the function.

The following is an example of using the create command to deploy a function by applying a customized SerDe.

 bin/pulsar-admin functions create \
  --jar /path/to/your.jar \
  --output-serde-classname com.example.serde.TweetSerde \
  # Other function attributes

note

Custom SerDe classes must be packaged with your function JARs.

Use SerDe for Python functions

In Python, the default SerDe is an identity, meaning that the type is serialized as whatever type the function returns.

For example, you can specify the SerDe as follows when deploying a function in cluster mode.

bin/pulsar-admin functions create \
  --tenant public \
  --namespace default \
  --name my_function \
  --py my_function.py \
  --classname my_function.MyFunction \
  --custom-serde-inputs '{"input-topic-1":"Serde1","input-topic-2":"Serde2"}' \
  --output-serde-classname Serde3 \
  --output output-topic-1

This case contains two input topics: input-topic-1 and input-topic-2, each of which is mapped to a different SerDe class (the mapping must be specified as a JSON string). The output topic output-topic-1 uses the Serde3 class for SerDe.

note

All function-related logic, including processing and SerDe classes, must be contained within a single Python file.

The table outlines three SerDe options for Python functions.

For example, imagine that you are writing a function that processes tweet objects. You can refer to the following example of the Tweet class in Python.

class Tweet(object):
    def __init__(self, username, tweet_content):
        self.username = username
        self.tweet_content = tweet_content

To use this class in Pulsar Functions, you have two options:

• Specify PickleSerDe, which applies the pickle library for SerDe.
• Create your own SerDe class. The following is an example.

from pulsar import SerDe
class TweetSerDe(SerDe):
    def serialize(self, input):
        return bytes("{0}|{1}".format(input.username, input.tweet_content))
    def deserialize(self, input_bytes):
        tweet_components = str(input_bytes).split('|')
        return Tweet(tweet_components[0], tweet_componentsp[1])

For more details, see code example.