Kafka Connector Specifications

The Kafka connector is a Java^® program that requires at least Java 8. To use the Kafka connector, the JAVA_HOME environment variable on your server machine must be set to the path of your Java 8 installation.

Each deployed archive that contains a streaming analytic function requires its own Kafka connector. For example, if you have two archives, you require two connectors. You do not have to install the Kafka connector twice, but you must run it twice and have exactly one Kafka connector configuration file per archive.

The lifecycle management of the Kafka connector depends on your production environment. Streaming Data Framework for MATLAB Production Server provides tools to make starting, stopping, and controlling the Kafka connector easier.

Prerequisites for Running Example

The following example provides a sample stateful streaming analytic function, shows how to package and deploy it to MATLAB Production Server, and shows how to manage the Kafka connector on the server.

To run the example, you require sample streaming data and a running MATLAB Production Server instance with a running persistence service.

Write Streaming Analytic MATLAB Function

For this example, use the sample MATLAB functions recamanSum and initRecamanSum, which are located in the support_package_root\toolbox\mps\streaming\Examples\Numeric folder, where support_package_root is the root folder of support packages on your system. To get the path to this folder, use this command:

fullfile(matlabshared.supportpkg.getSupportPackageRoot,'toolbox','mps','streaming','Examples','Numeric')

Later, you package the recamanSum function and deploy it to MATLAB Production Server. For details about the recamanSum and initRecamanSum functions and to access the code, see Write Stateful Function and Write State Initialization Function.

Package Streaming Analytic Function

To package the recamanSum streaming analytic into a deployable archive, you can run the following script. The script creates an input KafkaStream object dataKS connected to the recamanSum_data topic and an output KafkaStream object resultKS connected to the recamanSum_results topic. Then, the script uses the streamingDataCompiler function to launch the Production Server Compiler (MATLAB Compiler SDK) app. Using the app, you create a deployable archive recamanSum.ctf suitable for deployment to MATLAB Production Server. Provide the StateStore input argument in the call to streamingDatacompiler and set its value to RR. RR is the persistence connection name you created in Start Persistence Service.

kafkaHost = "kafka.host.com";
kafkaPort = 9092;

dataKS = kafkaStream(kafkaHost, kafkaPort, "recamanSum_data", Rows=100);

resultKS = kafkaStream(kafkaHost, kafkaPort, "recamanSum_results", ...
    Rows=100);

archive = streamingDataCompiler("recamanSum", dataKS, resultKS, ...
    InitialState="initRecamanSum", StateStore="RR");

From the Production Server Compiler app, click Package to create the recamanSum archive. When the packaging process finishes, open the output folder. In the output folder, navigate to the for_distribution folder. The for_distribution folder contains the recamanSum.ctf deployable archive and Kafka connector scripts that you use later.

Deploy Streaming Analytic Function to Server

Deploy the recamanSum archive to a running MATLAB Production Server instance. If you manage the server using the command line, copy the recamanSum archive to the auto_deploy folder of your server instance. For other ways to deploy, see Deploy Archive to MATLAB Production Server.

Start Kafka Connector

Depending on the operating system of your server instance, enter the following commands at the system prompt to start the Kafka connector script kafka-connector-start. The Kafka connector pulls data from the Kafka host and pushes it to the deployed streaming archive.

The output of the start script is a process ID (PID). Save the value of the PID. You use this ID to stop the Kafka Connector process later.

powershell -executionPolicy bypass -File kafka-connector-start.ps1 -out out.log -err error.log -c collector.properties -k kafka.properties

 chmod +x kafka-connector-start.sh

./kafka-connector-start.sh -out out.log -err error.log -c collector.properties -k kafka.properties

Read Processed Data from Output Stream

After you start the Kafka Connector, the server starts receiving several requests. The deployed recamanSum archive receives streaming data from the input Kafka stream as input and calculates the cumulative sum of the Recamán sequence. Wait a few seconds for the server to finish processing these requests.

Create another KafkaStream object to read the result from the output topic.

readStream = kafkaStream("kafka.host.com", 9092, "recamanSum_results");

Call readtimetable to read the output data.

result = readtimetable(readStream)

Stop Kafka Connector

Depending on the operating system of your server instance, enter the following commands at the system prompt to stop the Kafka Connector script kafka-connector-stop. Replace PID with the process ID that you save when you start the connector.

powershell -executionPolicy bypass -File kafka-connector-stop.ps1 PID

 chmod +x kafka-connector-stop.sh

 ./kafka-connector-stop.sh PID