How to create and validate ksqlDB UDFs

Table of Contents

In this article we will go over the process to create and validate ksqlDB UDFs.


  • Install Confluent platform for your operating system using instructions at “”
  • Install maven sudo apt install maven

Steps to create ksqlDB UDFs

  • Once we have the prerequisites, Lets go ahead and add the maven repositories from Confluent to your ~/.m2/settings.xml file

                <!-- Confluent releases -->  
                <!-- further repository entries here -->
  • Run the following command and fill in required details. Make sure to use the right profile if you have multiple maven profiles by using -Pmyprofile parameter

    mvn -Pmyprofile archetype:generate -X \
       -DarchetypeGroupId=io.confluent.ksql \
       -DarchetypeArtifactId=ksql-udf-quickstart \

  • Define value for property ‘groupId’: com.entechlog.ksql.functions
  • Define value for property ‘artifactId’: ksqlDB-udf
  • Define value for property ‘version’: 0.1.0
  • Define value for property ‘package’: com.entechlog.ksql.functions
  • Define value for property ‘author’: Siva Nadesan

  • This will create a new project with the following content in it

    │   pom.xml
        │   ├───java
        │   │   └───com
        │   │       └───entechlog
        │   │           └───ksql
        │   │               └───functions
        │   │             
        │   │             
        │   │
        │   └───resources
  • Update pom.xml for required dependencies. The archetype from Confluent includes an example UDF and UDAF, You can use that for testing to make sure your maven setup is correct by issuing the command mvn clean package inside the root project folder

  • Here I am going to create a custom KSQL function, which will return a capitalized version of the first character of each word in a string. Java code for this function can be found at “", Save this as “” in the same path as “”

  • Build the project by running the following command in the project root directory

    mvn clean package

  • Now copy the jar(udf-ToTitleCase-0.1.0.jar) to “ksql.extension.dir”

    sudo cp udf-ToTitleCase-0.1.0.jar /etc/ksql/ext
  • If you don’t have “ksql.extension.dir”, set the same by editing “”

    nano /etc/ksql/
  • After the copy (and properties change), restart ksql server based on your installation type

    confluent stop ksql-server;
    confluent start ksql-server;
    confluent status ksql-server; 
    sudo systemctl stop confluent-ksql.service;
    sudo systemctl start confluent-ksql.service;
    sudo systemctl status confluent-ksql.service;
  • Validate the properties and function in a ksql session using below commands

  • Now to test this, let’s set up some test data using ksql-datagen. If you don’t have “ksql-datagen”, install the same using the below command

    sudo confluent-hub install confluentinc/kafka-connect-datagen:latest
  • Create a schema for your datagen, See “" for the schema I have created to generate this test data

  • cd into the directory which has the schema (in this case user.avro) and issue below command. This will generate test data for our topic

    ksql-datagen schema=user.avro format=avro topic='dev.etl.datagen.user.src.0020' key=userid maxInterval=5000 iterations=10

  • Now, Start KSQL session and Issue this command to start reading data from the beginning of the topic

    SET 'auto.offset.reset'='earliest';
  • Check messages in topic using

    ksql> PRINT 'dev.etl.datagen.user.src.0020' FROM BEGINNING LIMIT 1;
    2/22/20 3:24:07 PM EST, 1000, {"userid": "1000", "fullname": "agustin waters", "firstname": "Brice", "lastname": "Dougherty", "countrycode": "AU"}
  • Now lets create a stream on this topic using

    CREATE STREAM STM_DATAGEN_USER_SRC_0010 WITH (KAFKA_TOPIC='dev.etl.datagen.user.src.0020', VALUE_FORMAT='Avro');
  • Validate the function now using below PUSH query. As you see here the function is working as expected

    ksql> SELECT userid, fullname, UDF_TOTITLECASE(fullname) FROM STM_DATAGEN_USER_SRC_0010 EMIT CHANGES;
    |USERID |FULLNAME                  |KSQL_COL_2           |
    |1000   |agustin waters            |Agustin Waters       |
    |1001   |tameka velasquez          |Tameka Velasquez     |
    |1002   |bruno ponce               |Bruno Ponce          |

Hope this was helpful. I will keep my github page “" updated as I explore and create more ksqlDB UDF’s.


Share this blog:

Related Articles