Pulsar Functions are lightweight compute processes that
Functions can be managed via the following methods.
Method | Description |
---|---|
Admin CLI | The functions command of the pulsar-admin tool. |
REST API | The /admin/v3/functions endpoint of the admin {@inject: rest:REST:/} API. |
Java Admin API | The functions method of the {@inject: javadoc:PulsarAdmin:/admin/org/apache/pulsar/client/admin/PulsarAdmin} object in the Java API. |
You can perform the following operations on functions.
You can create a Pulsar function in cluster mode (deploy it on a Pulsar cluster) using Admin CLI, REST API or Java Admin API.
Use the create
subcommand.
Example
$ pulsar-admin functions create \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --inputs test-input-topic \ --output persistent://public/default/test-output-topic \ --classname org.apache.pulsar.functions.api.examples.ExclamationFunction \ --jar /examples/api-examples.jar
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName?version=[[pulsar:version_number]]}
FunctionConfig functionConfig = new FunctionConfig(); functionConfig.setTenant(tenant); functionConfig.setNamespace(namespace); functionConfig.setName(functionName); functionConfig.setRuntime(FunctionConfig.Runtime.JAVA); functionConfig.setParallelism(1); functionConfig.setClassName("org.apache.pulsar.functions.api.examples.ExclamationFunction"); functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); functionConfig.setTopicsPattern(sourceTopicPattern); functionConfig.setSubName(subscriptionName); functionConfig.setAutoAck(true); functionConfig.setOutput(sinkTopic); admin.functions().createFunction(functionConfig, fileName);
You can update a Pulsar function that has been deployed to a Pulsar cluster using Admin CLI, REST API or Java Admin API.
Use the update
subcommand.
Example
$ pulsar-admin functions update \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --output persistent://public/default/update-output-topic \ # other options
{@inject: endpoint|PUT|/admin/v3/functions/:tenant/:namespace/:functionName?version=[[pulsar:version_number]]}
FunctionConfig functionConfig = new FunctionConfig(); functionConfig.setTenant(tenant); functionConfig.setNamespace(namespace); functionConfig.setName(functionName); functionConfig.setRuntime(FunctionConfig.Runtime.JAVA); functionConfig.setParallelism(1); functionConfig.setClassName("org.apache.pulsar.functions.api.examples.ExclamationFunction"); UpdateOptions updateOptions = new UpdateOptions(); updateOptions.setUpdateAuthData(updateAuthData); admin.functions().updateFunction(functionConfig, userCodeFile, updateOptions);
You can start a stopped function instance with instance-id
using Admin CLI, REST API or Java Admin API.
Use the start
subcommand.
$ pulsar-admin functions start \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --instance-id 1
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/:instanceId/start?version=[[pulsar:version_number]]}
admin.functions().startFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
You can start all stopped function instances using Admin CLI, REST API or Java Admin API.
Use the start
subcommand.
Example
$ pulsar-admin functions start \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/start?version=[[pulsar:version_number]]}
admin.functions().startFunction(tenant, namespace, functionName);
You can stop a function instance with instance-id
using Admin CLI, REST API or Java Admin API.
Use the stop
subcommand.
Example
$ pulsar-admin functions stop \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --instance-id 1
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/:instanceId/stop?version=[[pulsar:version_number]]}
admin.functions().stopFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
You can stop all function instances using Admin CLI, REST API or Java Admin API.
Use the stop
subcommand.
Example
$ pulsar-admin functions stop \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/stop?version=[[pulsar:version_number]]}
admin.functions().stopFunction(tenant, namespace, functionName);
Restart a function instance with instance-id
using Admin CLI, REST API or Java Admin API.
Use the restart
subcommand.
Example
$ pulsar-admin functions restart \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --instance-id 1
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/:instanceId/restart?version=[[pulsar:version_number]]}
admin.functions().restartFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
You can restart all function instances using Admin CLI, REST API or Java admin API.
Use the restart
subcommand.
Example
$ pulsar-admin functions restart \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/restart?version=[[pulsar:version_number]]}
admin.functions().restartFunction(tenant, namespace, functionName);
You can list all Pulsar functions running under a specific tenant and namespace using Admin CLI, REST API or Java Admin API.
Use the list
subcommand.
Example
$ pulsar-admin functions list \ --tenant public \ --namespace default
{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace?version=[[pulsar:version_number]]}
admin.functions().getFunctions(tenant, namespace);
You can delete a Pulsar function that is running on a Pulsar cluster using Admin CLI, REST API or Java Admin API.
Use the delete
subcommand.
Example
$ pulsar-admin functions delete \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions)
{@inject: endpoint|DELETE|/admin/v3/functions/:tenant/:namespace/:functionName?version=[[pulsar:version_number]]}
admin.functions().deleteFunction(tenant, namespace, functionName);
You can get information about a Pulsar function currently running in cluster mode using Admin CLI, REST API or Java Admin API.
Use the get
subcommand.
Example
$ pulsar-admin functions get \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions)
{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/:functionName?version=[[pulsar:version_number]]}
admin.functions().getFunction(tenant, namespace, functionName);
You can get the current status of a Pulsar function instance with instance-id
using Admin CLI, REST API or Java Admin API.
Use the status
subcommand.
Example
$ pulsar-admin functions status \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --instance-id 1
{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/:functionName/:instanceId/status?version=[[pulsar:version_number]]}
admin.functions().getFunctionStatus(tenant, namespace, functionName, Integer.parseInt(instanceId));
You can get the current status of a Pulsar function instance using Admin CLI, REST API or Java Admin API.
Use the status
subcommand.
Example
$ pulsar-admin functions status \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions)
{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/:functionName/status?version=[[pulsar:version_number]]}
admin.functions().getFunctionStatus(tenant, namespace, functionName);
You can get the current stats of a Pulsar Function instance with instance-id
using Admin CLI, REST API or Java admin API.
Use the stats
subcommand.
Example
$ pulsar-admin functions stats \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --instance-id 1
{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/:functionName/:instanceId/stats?version=[[pulsar:version_number]]}
admin.functions().getFunctionStats(tenant, namespace, functionName, Integer.parseInt(instanceId));
You can get the current stats of a Pulsar function using Admin CLI, REST API or Java admin API.
Use the stats
subcommand.
Example
$ pulsar-admin functions stats \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions)
{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/:functionName/stats?version=[[pulsar:version_number]]}
admin.functions().getFunctionStats(tenant, namespace, functionName);
You can trigger a specified Pulsar function with a supplied value using Admin CLI, REST API or Java admin API.
Use the trigger
subcommand.
Example
$ pulsar-admin functions trigger \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --topic (the name of input topic) \ --trigger-value \"hello pulsar\" # or --trigger-file (the path of trigger file)
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/trigger?version=[[pulsar:version_number]]}
admin.functions().triggerFunction(tenant, namespace, functionName, topic, triggerValue, triggerFile);
You can put the state associated with a Pulsar function using Admin CLI, REST API or Java admin API.
Use the putstate
subcommand.
Example
$ pulsar-admin functions putstate \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --state "{\"key\":\"pulsar\", \"stringValue\":\"hello pulsar\"}"
{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/:functionName/state/:key?version=[[pulsar:version_number]]}
TypeReference<FunctionState> typeRef = new TypeReference<FunctionState>() {}; FunctionState stateRepr = ObjectMapperFactory.getThreadLocal().readValue(state, typeRef); admin.functions().putFunctionState(tenant, namespace, functionName, stateRepr);
You can fetch the current state associated with a Pulsar function using Admin CLI, REST API or Java admin API.
Use the querystate
subcommand.
Example
$ pulsar-admin functions querystate \ --tenant public \ --namespace default \ --name (the name of Pulsar Functions) \ --key (the key of state)
{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/:functionName/state/:key?version=[[pulsar:version_number]]}
admin.functions().getFunctionState(tenant, namespace, functionName, key);