We will use MNIST data sample. The training sample contains 60,000 images with handwritten digits from 0 to 9 and the same amount of labels with corresponding digits. The test sample contains 1,000 images. The samples are available as gzip binaries. As we assume a use case where the data of interest are in the database, we need to load the data into the database first.

We need to generate two SQL files, one for the train data set, mnist_train.sql, and one for a test data sample, mnist_test.sql. Each file would contain SQL commands for creating a table with two columns: pixels column of type BYTEA, which would contain raw image bytes, and labels column of type SMALLINT, which would contain digit corresponding to the image in given table row. The rest of the file would contain commands for populating the table. Image bytes can be decoded as a HEX string.

As we will show how to leverage Debezium for data streaming later in this post, we will initially load the training data set into the database. The SQL file with training data will be used directly by the Postgres container - when it starts, it will load this data into the training table. We will use the test data SQL file later. However, the preparation of the data is the same for training as well as test samples, and we can prepare both of them in one go.

To prepare these SQL files, you can use mnist2sql.py script from Debezium tensorflow-mnist example:

The script assumes MNIST data sets are available in the postgres directory. When using the --download parameter, the script first downloads MNIST data samples into the postgres directory. The postgres directory will contain the resulting SQL files.

The most common Debezium usage is the streaming of records to Kafka. TensorFlow provides TensorFlow I/O module for loading data from various sources. Besides other sources, it also allows loading the data from Kafka. There are several ways to do it. IODataset.from_kafka() method loads only existing data from specified Kafka topics. Two experimental classes support streaming data, KafkaBatchIODataset and KafkaGroupIODataset. Both are very similar and allow them to work with streaming data, i.e., they not only read the existing data from a Kafka topic but also wait for new data and eventually pass new records into the TensorFlow. Streaming concludes when there are no new events within a specified time frame.

In all cases, a Dataset represents all loaded records in Tensorflow. This Tensorflow data structure provides convenience for building data pipelines, which may include further data transformations or preprocessing.

This sounds great. However, the most significant caveat is the representation of records within the Dataset. These Kafka loaders completely ignore the schema of the records provided by Kafka, meaning that keys and values are raw bytes of data. Additionally, the ingestion pipeline complicates the process by converting these into strings (i.e., toString() on the object called). So if you pass, e.g., raw image bytes via Kafka, using Kafka BYTES_SCHEMA, it would result in something like this:

Instead of getting a batch of raw image bytes which you can further transform in TensorFlow, you get only string representation of Java byte arrays, which is not very useful.

The most straightforward solution would be to convert the raw image bytes into numbers before sending them to Kafka to mitigate the problem. As TensorFlow provides methods for parsing CSV input, we can convert each image into one CSV line of numbers. Since Tensorflow primarily works with numbers, we would be required to convert the images to numbers regardless. We can pass the number on the image as a message key. Now, a single message transform supported by Debezium comes in handy. The transformation can look like this:

On the TensorFlow side, we must convert bytes obtained from Kafka messages into numbers. The following illustrates a map function to handle this easily:

Here we parse CSV lines, potentially provided as the raw bytes, and immediately scale the numbers within the <0, 1> interval, which is convenient for training our model later. Loading the data and creating data batches is very straightforward:

Here we use IODataset.from_kafka() for loading existing data from the Kafka topic, use our map function to convert bytes into numbers, and scale the numbers. As a last step, we create batches from the data set for more efficient processing. Parameters of tfio.IODataset.from_kafka() are self-explanatory and probably don’t need further comments.

As a result, we have a data set formed by two-dimensional tensors. The first dimension is a vector of floats representing the image, while the second dimension is a single number (scalar) describing the number on the picture. Once we have prepared our training data set, we can define our neural network model.

To keep things simple, as the main goal of this post is not to show the best handwritten digit classifier, but to show how to create the data pipeline, let’s use a very simple model:

This model contains only two layers. Although this model is really simple, it still does a pretty good job in recognition of handwritten digits. Probably more interesting than the model itself is how easy it is to write a mode in TensorFlow (or actually Keras, but it’s now part of TensorFlow).

Similarly easy is to define model optimizer and the loss function:

It’s outside of this post’s scope to explain these functions, and you can check almost any machine learning online course or textbook on this topic for a detailed explanation.

Once we have our model ready, we can train it on the trained dataset prepared in the previous section:

This step may take quite some time to finish. However, once finished, our model is ready to recognize handwritten digits!

Let’s see how good our model is in digit recognition. But as our primary goal here is to explore the means how to ingest data into TensorFlow, we will start model evaluation on an empty (or, more accurately, even non-existing) Kafka topic and see if we will be able to evaluate the data on the fly as they will pop-up first in the database and then in the corresponding Kafka topic. For this purpose, we can use one of the streaming classes mentioned above:

Again, arguments are mostly self-explanatory. Two things may need further explanation: stream_timeout and configuration parameters. stream_timeout determines the interval of inactivity (in milliseconds) after which the streaming would terminate. configuration is librdkafka configuration. It’s a configuration of the Kafka client; you should configure at least the session timeout (session.timeout.ms), and it’s poll interval (max.poll.interval.ms). The values of these parameters should be higher than the value of stream_timeout.

The dataset this loader provides is slightly different - instead of providing a single record containing the message and its key, we get the key and message already split. Therefore, we have to define a slightly modified map function with two arguments:

With this function, we can adjust the dataset and create batches as before:

and evaluate the model:

You can execute a cell with model evaluation in the Jupyter notebook. The execution will wait because there is no such topic in Kafka and no table with test data in the database. The streaming timeout is 9 seconds, so data must be provided within this time frame after launching the model evaluation. At the start of this demo, we created a SQL file in the postgres directory called mnist_test.sql, which can generate the test data we need:

After a short while, you should see in the Jupyter notebook output that some data arrived into the model and, a few moments later final evaluation of the model.

To make the results closer to humans, let’s define an image manually and serve it to the model. We can also easily show the image in the Jypiter notebook. The function for plotting the images and providing model predictions as a plot title can look like this:

Probably the only cryptic line in this function is the one containing the softmax() function. This function converts the resulting vector into a vector of probabilities. Elements of this vector express the probability that the number on a given position is the one on the image. Therefore, the position with the highest probability is the model’s prediction, where argmax() is derived.

We can try it, e.g., for this image, which contains the handwritten number 3:

The result would be as follows:

You can do the same by reading from a Kafka stream, and we can reuse existing topics for this purpose. As we already read all records from the test stream, we need to change the Kafka consumer group if we want to reread it using streaming KafkaGroupIODataset:

If you want to create a new stream and verify that our model can provide prediction as the new data arrives, you can easily do so:

In such case you don’t need to change Kafka consumer group, but you have to change the Kafka topic:

In either case, the result should look like this: