- Process Data Streams
- Develop Functions
Develop Pulsar Functions in WASM(Private Preview)
This section introduces how to develop and pacakge WASM Pulsar functions to use on StreamNative cloud.
The WASM runtime is still in private preview stage, If you want to try it out or have any questions, please submit a ticket to the support team.
Develop
The WASM runtime is using WasmEdge, theoretically, you can use any languages which can be compiled to a WASM module to write your functions, below is an example using Rust:
- cargo.toml
[package]
name = "excla"
version = "0.1.0"
edition = "2021"
[lib]
crate-type = ["cdylib"]
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
apache-avro = { version = "0.15.0", features = ["bzip", "xz", "snappy", "zstandard"] }
serde = "^1.0"
serde_json = "^1.0"
wasmedge-bindgen = "0.4.1"
wasmedge-bindgen-macro = "0.4.1"
- src/lib.rs
#[allow(unused_imports)]
use wasmedge_bindgen::*;
use wasmedge_bindgen_macro::*;
use serde::{Deserialize, Serialize};
#[derive(Debug, Deserialize, Serialize)]
pub struct Student {
pub name: Option<String>,
pub age: Option<i32>,
pub grade: Option<i32>
}
// use `process_json` as the class name when creating functions
#[wasmedge_bindgen]
pub fn process_json(s: Vec<u8>) -> Vec<u8> {
let stu = serde_json::from_slice::<Student>(&s[..]).unwrap();
let stu = Student {
grade: stu.grade.map(|grade| grade + 1),
..stu
};
let res = serde_json::to_vec(&stu).unwrap();
return res
}
Feature Matrix
The WASM runtime doesn't support full features comparing to Java runtime, and it's still in developing, below is the matrix:
Input Arguments
Input | Java | Go(Pulsar) | Python | WASM |
---|---|---|---|---|
Custom SerDe | ✅ | ❌ | ✅ | ? |
Schema - Avro | ✅ | ❌ | ✅ | ? |
Schema - JSON | ✅ | ❌ | ✅ | ? |
Schema - Protobuf | ✅ | ❌ | ❌ | ? |
Schema - KeyValue | ✅ | ❌ | ❌ | ? |
Schema - AutoSchema | ✅ | ❌ | ❌ | ? |
Scehma - Protobuf Native | ✅ | ❌ | ❌ | ? |
e-2-e encryption | ✅ | ❌ | ✅ | ✅ |
maxMessageRetries | ✅ | ❌ | ❌ | ✅ |
dead-letter policy | ✅ | ❌ | ❌ | ✅ |
SubscriptionName | ✅ | ✅ | ✅ | ✅ |
SubscriptionType | ✅ | ✅ | ✅ | ✅ |
SubscriptionInitialPosition | ✅ | ❌ | ✅ | ✅ |
AutoAck | ✅ | ✅ | ✅ | ✅ |
Note
Users can implement the Schema themselves since we are passing and expecting []byte to/from the users' function, so leave ? here.
Output Arguments
Output | Java | Go(Pulsar) | Python | WASM |
---|---|---|---|---|
Custom SerDe | ✅ | ❌ | ✅ | ? |
Schema - Avro | ✅ | ❌ | ✅ | ? |
Schema - JSON | ✅ | ❌ | ✅ | ? |
Schema - Protobuf | ✅ | ❌ | ❌ | ? |
Schema - KeyValue | ✅ | ❌ | ❌ | ? |
Schema - AutoSchema | ✅ | ❌ | ❌ | ? |
Schema - Protobuf Native | ✅ | ❌ | ❌ | ? |
useThreadLocalProducers | ✅ | ❌ | ❌ | ✅ |
Key-based Batcher | ✅ | ✅ | ✅ | ✅ |
e-2-e encryption | ✅ | ❌ | ✅ | ✅ |
Compression | ✅ | ✅ | ✅ | ✅ |
Context
WASM runtime doesn't support the Context features at all for now.
Other
Other | Java | Go(Pulsar) | Python | WASM |
---|---|---|---|---|
Resources | ✅ | ✅ | ✅ | ✅ |
At-most-once | ✅ | ✅ | ✅ | ✅ |
At-least-once | ✅ | ✅ | ✅ | ✅ |
Effectively-once | ✅ | ❌ | ✅ | ❌ |
Package
You need to compile the function to a .wasm
module first before creating Pulsar Functions.
cargo build --target wasm32-wasi --release
Deploy
After creating a cluster, set up your environment and develop&package your function, you can use the snctl
, pulsarctl
, pulsar-admin
command, the REST API, or terraform
to deploy a Pulsar function to your cluster.
You can create a WASM Pulsar function by using a local .wasm
file or an uploaded Pulsar functions package(recommend).
(Optional) Upload your function file to Pulsar
It's recommend to upload your function file to Pulsar before you create a function. Since you can add a version suffix to the package.
Upload packages
snctl pulsar admin packages upload function://public/default/[email protected] \
--path exclamation-1.0-SNAPSHOT.jar \
--description "exclamation function" \
--properties fileName=exclamation-1.0-SNAPSHOT.jar
You should see the following output:
The package 'function://public/default/[email protected]' uploaded from path 'examples/api-examples.jar' successfully
Create
snctl pulsar admin functions create \
--tenant public \
--namespace default \
--name function1 \
--inputs persistent://public/default/test-wasm-input \
--output persistent://public/default/test-wasm-output \
--classname exclamation \
--py function://public/default/[email protected] \
--custom-runtime-options '{"genericKind": "wasm"}'
--use-service-account
Note
Since Pulsar doesn't support WASM runtime, we need to use --py
to specify the function file and specify the --custom-runtime-options '{"genericKind": "wasm"}'
to make it work.
You should see something like this:
Created function1 successfully
For details about Pulsar function configurations, see Pulsar function configurations.
What’s next?
- Learn how to manage functions.
- Learn how to monitor functions.
- Reference common configurations.