Apache Flink has excellent integration with Kafka. We can pass the Debezium records as e.g. JSON records. For creating Flink tables, it even has support for Debezium’s record format, but for streams, we need to extract part of the Debezium message, which contains the newly stored row of the table. However, this is very easy as Debezium provides SMT, extract new record state SMT, which does precisely this. The complete Debezium configuration can look like this:

The configuration captures all tables in the public schema with tables that begin with the iris_ prefix. Since we are storing training and test data in two tables, two Kafka topics named flink.public.iris_train and flink.public.iris_test are created, respectively. Flink’s DataStreamSource represents the incoming stream of data. As we encode the records as a JSON, it will be a stream of JSON ObjectNode objects. Constructing the source stream is very straightforward:

Flink operates primarily on the Table abstraction object. Also, ML models accept only tables as input, and predictions are produced as tables too. Therefore, we must first convert our input stream into a Table object. We will start by transforming our input data stream into a stream of table rows. We need to define a map function that would return a Row object with a vector containing one data point. As the k-means algorithm belongs to unsupervised learning algorithms, i.e. the model doesn’t need corresponding "right answers" for the data points, we can skip the label field from the vector:

Various parts of the internal Flink pipeline can run on different worker nodes, and therefore, we also need to provide type information about the table. With that, we are ready to create the table object:

Once we have a Table object, we can pass it to our model. So let’s create one and pass a train stream to it for continuous model training:

To make things more straightforward, we directly set the number of desired clusters to 3 instead of finding the optimal number of clusters by digging into the data (using e.g. elbow method). We also set some initial values for the centers of the clusters instead of using random numbers (Flink provides a convenient method for it - KMeansModelData.generateRandomModelData() if you want to try with random centers).

To obtain the predictions for our test data, we again need to convert our test stream into a table. The model transforms the table with test data into a table with predictions. Finally, convert the prediction into a stream and persisted, e.g. in a Kafka topic:

Now, we are ready to build our application and almost ready to submit it to Flink for execution. Before we do, we need to create the required Kafka topics first. While the topics can be empty, Flink requires that they at least exist. As we include a small set of data in the Postgres training table when the database starts, Debezium will create a corresponding topic when registering the Debezium Postgres connector in Kafka Connect. Since the test data table does not yet exist, we need to create the topic in Kafka manually:

Now, we are ready to submit our application to Flink. For the complete code, please see the corresponding source code in Debezium example repository

Flink provides a friendly UI, which is available on http://localhost:8081/. There, you can check, besides other things, the status of your jobs and also, e.g. job execution plan in an excellent graphical representation:

From the user’s point of view, all the interactions with our model occur by inserting new records into the database or reading Kafka topics with predictions. As we already created a very small initial training data sample in the database when it started, we can directly check our model predictions by inserting our test data sample into the database:

The insert results in an immediate data stream of test data in Kafka, passing it into the model and sending the prediction back to the iris_predictions Kafka topic. The predictions are not accurate when training the model on a very small data set with just two clusters. The following shows our initial predictions:

In our case, the correct answer should be:

When comparing the result, we only have 5 of 11 data points correctly classified due to the initial sample training data size. On the other hand, as we didn’t start with completely random clusters, our predictions are also not completely wrong.

Let’s see how things change when we supply more training data into the model:

To see the updated predictions, we insert the same test data sample again into the database:

The following predictions are much better since we have all three categories present. We have also correctly classified 7 out of the 11 data points.

As the whole data sample is pretty small, for further model training we can re-use our second train data sample:

This results in the following prediction.

We now find we have 9 out of 11 data points correctly classified. While this is still not an excellent result, we expect only partially accurate results as this is simply a prediction. The primary motivation here is to show the whole pipeline and demonstrate that the model improves the predictions without re-training and re-deploying the model when adding new data.

Similarly to Flink, creating Spark streams from Kafka streams is straightforward, and most of the parameters are self-explanatory:

On the last line, we transform the Kafka stream to a labeled point stream, which the Spark ML library uses for working with its ML models. Labeled points are expected as the strings formatted as data point labels separated by the comma from space-separated data point values. So the map function looks like this:

It still applies that k-means is an unsupervised algorithm and doesn’t use the data point labels. However, it’s convenient to pass them to LabeledPoint class as later on, we can show them together with model predictions.

We chain one more map function to parse the string and create a labeled data point from it. In this case, it’s a built-in function of Spark LabeledPoint.

Contrary to Flink, Spark doesn’t require Kafka topics to exist in advance, so when deploying the model, we don’t have to create the topics. We can let Debezium create them once the table with the test data is created and populated with the data.

Defining the streaming k-means model is very similar to Flink:

Also, in this case, we directly set the number of clusters to 3 and provide the same initial central points to the clusters. We also only pass the data points for training, not the labels.

As mentioned above, we can use the labels to show them together with the predictions:

We print 11 stream elements to the console on the resulting stream with the predictions, as this is the size of our test sample. Like Flink, the results after initial training on a very small data sample could be better. The first number in the tuple is the data point label, while the second one is the corresponding prediction done by our model:

However, when we provide more training data, predictions are much better:

If we pass the second training data sample once again for the training, our model makes correct predictions for the whole test sample:

So, similar to Flink, we get better results as we pass more training data without the need to re-train and re-deploy the model. In this case, we get even better results than Flink’s model.