Robust machine learning on streaming data using Kafka and Tensorflow-IO

View on TensorFlow.org Run in Google Colab View source on GitHub Download notebook

Overview

This tutorial focuses on streaming data from a Kafka cluster into a tf.data.Dataset which is then used in conjunction with tf.keras for training and inference.

Kafka is primarily a distributed event-streaming platform which provides scalable and fault-tolerant streaming data across data pipelines. It is an essential technical component of a plethora of major enterprises where mission-critical data delivery is a primary requirement.

Setup

Install the required tensorflow-io and kafka packages

pip install -q tensorflow-io
pip install -q kafka-python

Import packages

import os
from datetime import datetime
import time
import threading
import json
from kafka import KafkaProducer
from kafka.errors import KafkaError
from sklearn.model_selection import train_test_split
import pandas as pd
import tensorflow as tf
import tensorflow_io as tfio

Validate tf and tfio imports

print("tensorflow-io version: {}".format(tfio.__version__))
print("tensorflow version: {}".format(tf.__version__))
tensorflow-io version: 0.17.0
tensorflow version: 2.4.1

Download and setup Kafka and Zookeeper instances

For demo purposes, the following instances are setup locally:

  • Kafka (Brokers: 127.0.0.1:9092)
  • Zookeeper (Node: 127.0.0.1:2181)
curl -sSOL https://downloads.apache.org/kafka/2.7.0/kafka_2.13-2.7.0.tgz
tar -xzf kafka_2.13-2.7.0.tgz

Using the default configurations (provided by Apache Kafka) for spinning up the instances.

./kafka_2.13-2.7.0/bin/zookeeper-server-start.sh -daemon ./kafka_2.13-2.7.0/config/zookeeper.properties
./kafka_2.13-2.7.0/bin/kafka-server-start.sh -daemon ./kafka_2.13-2.7.0/config/server.properties
echo "Waiting for 10 secs until kafka and zookeeper services are up and running"
sleep 10
Waiting for 10 secs until kafka and zookeeper services are up and running

Once the instances are started as daemon processes, grep for kafka in the processes list. The two java processes correspond to zookeeper and the kafka instances.

ps -ef | grep kafka
kbuilder  3128 22247  2 19:38 ?        00:00:00 python /tmpfs/src/gfile/executor.py --input_notebook=/tmpfs/src/temp/docs/tutorials/kafka.ipynb --timeout=15000
kbuilder  3562     1 13 19:38 ?        00:00:01 java -Xmx512M -Xms512M -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true -Xlog:gc*:file=/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../logs/zookeeper-gc.log:time,tags:filecount=10,filesize=100M -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dkafka.logs.dir=/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../logs -Dlog4j.configuration=file:./kafka_2.13-2.7.0/bin/../config/log4j.properties -cp /tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/activation-1.1.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/aopalliance-repackaged-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/argparse4j-0.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/audience-annotations-0.5.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/commons-cli-1.4.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/commons-lang3-3.8.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-api-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-basic-auth-extension-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-file-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-json-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-mirror-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-mirror-client-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-runtime-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-transforms-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/hk2-api-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/hk2-locator-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/hk2-utils-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-annotations-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-core-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-databind-2.10.5.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-dataformat-csv-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-datatype-jdk8-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-jaxrs-base-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-jaxrs-json-provider-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-module-jaxb-annotations-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-module-paranamer-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-module-scala_2.13-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.activation-api-1.2.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.annotation-api-1.3.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.inject-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.validation-api-2.0.2.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.ws.rs-api-2.1.6.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.xml.bind-api-2.3.2.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javassist-3.25.0-GA.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javassist-3.26.0-GA.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javax.servlet-api-3.1.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javax.ws.rs-api-2.1.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jaxb-api-2.3.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-client-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-common-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-container-servlet-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-container-servlet-core-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-hk2-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-media-jaxb-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-server-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-client-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-continuation-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-http-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-io-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-security-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-server-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-servlet-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-servlets-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-util-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jopt-simple-5.0.4.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-clients-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-log4j-appender-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-raft-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-examples-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-scala_2.13-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-test-utils-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-tools-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka_2.13-2.7.0-sources.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka_2.13-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/log4j-1.2.17.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/lz4-java-1.7.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/maven-artifact-3.6.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/metrics-core-2.2.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-buffer-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-codec-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-common-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-handler-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-resolver-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-transport-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-transport-native-epoll-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-transport-native-unix-common-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/osgi-resource-locator-1.0.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/paranamer-2.8.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/plexus-utils-3.2.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/reflections-0.9.12.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/rocksdbjni-5.18.4.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-collection-compat_2.13-2.2.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-java8-compat_2.13-0.9.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-library-2.13.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-logging_2.13-3.9.2.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-reflect-2.13.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/slf4j-api-1.7.30.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/slf4j-log4j12-1.7.30.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/snappy-java-1.1.7.7.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/zookeeper-3.5.8.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/zookeeper-jute-3.5.8.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/zstd-jni-1.4.5-6.jar org.apache.zookeeper.server.quorum.QuorumPeerMain ./kafka_2.13-2.7.0/config/zookeeper.properties
kbuilder  3930     1 52 19:38 ?        00:00:05 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true -Xlog:gc*:file=/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../logs/kafkaServer-gc.log:time,tags:filecount=10,filesize=100M -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dkafka.logs.dir=/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../logs -Dlog4j.configuration=file:./kafka_2.13-2.7.0/bin/../config/log4j.properties -cp /tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/activation-1.1.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/aopalliance-repackaged-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/argparse4j-0.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/audience-annotations-0.5.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/commons-cli-1.4.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/commons-lang3-3.8.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-api-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-basic-auth-extension-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-file-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-json-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-mirror-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-mirror-client-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-runtime-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/connect-transforms-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/hk2-api-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/hk2-locator-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/hk2-utils-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-annotations-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-core-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-databind-2.10.5.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-dataformat-csv-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-datatype-jdk8-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-jaxrs-base-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-jaxrs-json-provider-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-module-jaxb-annotations-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-module-paranamer-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jackson-module-scala_2.13-2.10.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.activation-api-1.2.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.annotation-api-1.3.5.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.inject-2.6.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.validation-api-2.0.2.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.ws.rs-api-2.1.6.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jakarta.xml.bind-api-2.3.2.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javassist-3.25.0-GA.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javassist-3.26.0-GA.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javax.servlet-api-3.1.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/javax.ws.rs-api-2.1.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jaxb-api-2.3.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-client-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-common-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-container-servlet-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-container-servlet-core-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-hk2-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-media-jaxb-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jersey-server-2.31.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-client-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-continuation-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-http-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-io-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-security-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-server-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-servlet-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-servlets-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jetty-util-9.4.33.v20201020.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/jopt-simple-5.0.4.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-clients-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-log4j-appender-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-raft-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-examples-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-scala_2.13-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-streams-test-utils-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka-tools-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka_2.13-2.7.0-sources.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/kafka_2.13-2.7.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/log4j-1.2.17.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/lz4-java-1.7.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/maven-artifact-3.6.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/metrics-core-2.2.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-buffer-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-codec-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-common-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-handler-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-resolver-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-transport-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-transport-native-epoll-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/netty-transport-native-unix-common-4.1.51.Final.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/osgi-resource-locator-1.0.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/paranamer-2.8.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/plexus-utils-3.2.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/reflections-0.9.12.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/rocksdbjni-5.18.4.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-collection-compat_2.13-2.2.0.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-java8-compat_2.13-0.9.1.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-library-2.13.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-logging_2.13-3.9.2.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/scala-reflect-2.13.3.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/slf4j-api-1.7.30.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/slf4j-log4j12-1.7.30.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/snappy-java-1.1.7.7.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/zookeeper-3.5.8.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/zookeeper-jute-3.5.8.jar:/tmpfs/src/temp/docs/tutorials/kafka_2.13-2.7.0/bin/../libs/zstd-jni-1.4.5-6.jar kafka.Kafka ./kafka_2.13-2.7.0/config/server.properties
kbuilder  4118  3132  0 19:38 pts/0    00:00:00 /bin/bash -c ps -ef | grep kafka
kbuilder  4120  4118  0 19:38 pts/0    00:00:00 grep kafka

Create the kafka topics with the following specs:

  • susy-train: partitions=1, replication-factor=1
  • susy-test: partitions=2, replication-factor=1
./kafka_2.13-2.7.0/bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --replication-factor 1 --partitions 1 --topic susy-train
./kafka_2.13-2.7.0/bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --replication-factor 1 --partitions 2 --topic susy-test
Created topic susy-train.
Created topic susy-test.

Describe the topic for details on the configuration

./kafka_2.13-2.7.0/bin/kafka-topics.sh --describe --bootstrap-server 127.0.0.1:9092 --topic susy-train
./kafka_2.13-2.7.0/bin/kafka-topics.sh --describe --bootstrap-server 127.0.0.1:9092 --topic susy-test
Topic: susy-train PartitionCount: 1 ReplicationFactor: 1  Configs: segment.bytes=1073741824
    Topic: susy-train Partition: 0  Leader: 0 Replicas: 0   Isr: 0
Topic: susy-test  PartitionCount: 2 ReplicationFactor: 1  Configs: segment.bytes=1073741824
    Topic: susy-test  Partition: 0  Leader: 0 Replicas: 0   Isr: 0
    Topic: susy-test  Partition: 1  Leader: 0 Replicas: 0   Isr: 0

The replication factor 1 indicates that the data is not being replicated. This is due to the presence of a single broker in our kafka setup. In production systems, the number of bootstrap servers can be in the range of 100's of nodes. That is where the fault-tolerance using replication comes into picture.

Please refer to the docs for more details.

SUSY Dataset

Kafka being an event streaming platform, enables data from various sources to be written into it. For instance:

  • Web traffic logs
  • Astronomical measurements
  • IoT sensor data
  • Product reviews and many more.

For the purpose of this tutorial, lets download the SUSY dataset and feed the data into kafka manually. The goal of this classification problem is to distinguish between a signal process which produces supersymmetric particles and a background process which does not.

curl -sSOL https://archive.ics.uci.edu/ml/machine-learning-databases/00279/SUSY.csv.gz

Explore the dataset

The first column is the class label (1 for signal, 0 for background), followed by the 18 features (8 low-level features then 10 high-level features). The first 8 features are kinematic properties measured by the particle detectors in the accelerator. The last 10 features are functions of the first 8 features. These are high-level features derived by physicists to help discriminate between the two classes.

COLUMNS = [
          #  labels
           'class',
          #  low-level features
           'lepton_1_pT',
           'lepton_1_eta',
           'lepton_1_phi',
           'lepton_2_pT',
           'lepton_2_eta',
           'lepton_2_phi',
           'missing_energy_magnitude',
           'missing_energy_phi',
          #  high-level derived features
           'MET_rel',
           'axial_MET',
           'M_R',
           'M_TR_2',
           'R',
           'MT2',
           'S_R',
           'M_Delta_R',
           'dPhi_r_b',
           'cos(theta_r1)'
           ]

The entire dataset consists of 5 million rows. However, for the purpose of this tutorial, let's consider only a fraction of the dataset (100,000 rows) so that less time is spent on the moving the data and more time on understanding the functionality of the api.

susy_iterator = pd.read_csv('SUSY.csv.gz', header=None, names=COLUMNS, chunksize=100000)
susy_df = next(susy_iterator)
susy_df.head()
# Number of datapoints and columns
len(susy_df), len(susy_df.columns)
(100000, 19)
# Number of datapoints belonging to each class (0: background noise, 1: signal)
len(susy_df[susy_df["class"]==0]), len(susy_df[susy_df["class"]==1])
(54025, 45975)

Split the dataset

train_df, test_df = train_test_split(susy_df, test_size=0.4, shuffle=True)
print("Number of training samples: ",len(train_df))
print("Number of testing sample: ",len(test_df))

x_train_df = train_df.drop(["class"], axis=1)
y_train_df = train_df["class"]

x_test_df = test_df.drop(["class"], axis=1)
y_test_df = test_df["class"]

# The labels are set as the kafka message keys so as to store data
# in multiple-partitions. Thus, enabling efficient data retrieval
# using the consumer groups.
x_train = list(filter(None, x_train_df.to_csv(index=False).split("\n")[1:]))
y_train = list(filter(None, y_train_df.to_csv(index=False).split("\n")[1:]))

x_test = list(filter(None, x_test_df.to_csv(index=False).split("\n")[1:]))
y_test = list(filter(None, y_test_df.to_csv(index=False).split("\n")[1:]))
Number of training samples:  60000
Number of testing sample:  40000
NUM_COLUMNS = len(x_train_df.columns)
len(x_train), len(y_train), len(x_test), len(y_test)
(60000, 60000, 40000, 40000)

Store the train and test data in kafka

Storing the data in kafka simulates an environment for continuous remote data retrieval for training and inference purposes.

def error_callback(exc):
    raise Exception('Error while sendig data to kafka: {0}'.format(str(exc)))

def write_to_kafka(topic_name, items):
  count=0
  producer = KafkaProducer(bootstrap_servers=['127.0.0.1:9092'])
  for message, key in items:
    producer.send(topic_name, key=key.encode('utf-8'), value=message.encode('utf-8')).add_errback(error_callback)
    count+=1
  producer.flush()
  print("Wrote {0} messages into topic: {1}".format(count, topic_name))

write_to_kafka("susy-train", zip(x_train, y_train))
write_to_kafka("susy-test", zip(x_test, y_test))
Wrote 60000 messages into topic: susy-train
Wrote 40000 messages into topic: susy-test

Define the tfio train dataset

The IODataset class is utilized for streaming data from kafka into tensorflow. The class inherits from tf.data.Dataset and thus has all the useful functionalities of tf.data.Dataset out of the box.

def decode_kafka_item(item):
  message = tf.io.decode_csv(item.message, [[0.0] for i in range(NUM_COLUMNS)])
  key = tf.strings.to_number(item.key)
  return (message, key)

BATCH_SIZE=64
SHUFFLE_BUFFER_SIZE=64
train_ds = tfio.IODataset.from_kafka('susy-train', partition=0, offset=0)
train_ds = train_ds.shuffle(buffer_size=SHUFFLE_BUFFER_SIZE)
train_ds = train_ds.map(decode_kafka_item)
train_ds = train_ds.batch(BATCH_SIZE)

Build and train the model

# Set the parameters

OPTIMIZER="adam"
LOSS=tf.keras.losses.BinaryCrossentropy(from_logits=True)
METRICS=['accuracy']
EPOCHS=10
# design/build the model
model = tf.keras.Sequential([
  tf.keras.layers.Input(shape=(NUM_COLUMNS,)),
  tf.keras.layers.Dense(128, activation='relu'),
  tf.keras.layers.Dropout(0.2),
  tf.keras.layers.Dense(256, activation='relu'),
  tf.keras.layers.Dropout(0.4),
  tf.keras.layers.Dense(128, activation='relu'),
  tf.keras.layers.Dropout(0.4),
  tf.keras.layers.Dense(1, activation='sigmoid')
])

print(model.summary())
Model: "sequential"
_________________________________________________________________
Layer (type)                 Output Shape              Param #   
=================================================================
dense (Dense)                (None, 128)               2432      
_________________________________________________________________
dropout (Dropout)            (None, 128)               0         
_________________________________________________________________
dense_1 (Dense)              (None, 256)               33024     
_________________________________________________________________
dropout_1 (Dropout)          (None, 256)               0         
_________________________________________________________________
dense_2 (Dense)              (None, 128)               32896     
_________________________________________________________________
dropout_2 (Dropout)          (None, 128)               0         
_________________________________________________________________
dense_3 (Dense)              (None, 1)                 129       
=================================================================
Total params: 68,481
Trainable params: 68,481
Non-trainable params: 0
_________________________________________________________________
None
# compile the model
model.compile(optimizer=OPTIMIZER, loss=LOSS, metrics=METRICS)
# fit the model
model.fit(train_ds, epochs=EPOCHS)
Epoch 1/10
938/938 [==============================] - 31s 32ms/step - loss: 0.5218 - accuracy: 0.7379
Epoch 2/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4578 - accuracy: 0.7858
Epoch 3/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4516 - accuracy: 0.7906
Epoch 4/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4492 - accuracy: 0.7908
Epoch 5/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4458 - accuracy: 0.7944
Epoch 6/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4466 - accuracy: 0.7947
Epoch 7/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4439 - accuracy: 0.7939
Epoch 8/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4442 - accuracy: 0.7941
Epoch 9/10
938/938 [==============================] - 31s 32ms/step - loss: 0.4421 - accuracy: 0.7951
Epoch 10/10
938/938 [==============================] - 30s 32ms/step - loss: 0.4423 - accuracy: 0.7962
<tensorflow.python.keras.callbacks.History at 0x7fe8b8405b00>

Since only a fraction of the dataset is being utilized, our accuracy is limited to ~78% during the training phase. However, please feel free to store additional data in kafka for a better model performance. Also, since the goal was to just demonstrate the functionality of the tfio kafka datasets, a smaller and less-complicated neural network was used. However, one can increase the complexity of the model, modify the learning strategy, tune hyper-parameters etc for exploration purposes. For a baseline approach, please refer to this article.

Infer on the test data

To infer on the test data by adhering to the 'exactly-once' semantics along with fault-tolerance, the streaming.KafkaGroupIODataset can be utilized.

Define the tfio test dataset

The stream_timeout parameter blocks for the given duration for new data points to be streamed into the topic. This removes the need for creating new datasets if the data is being streamed into the topic in an intermittent fashion.

test_ds = tfio.experimental.streaming.KafkaGroupIODataset(
    topics=["susy-test"],
    group_id="testcg",
    servers="127.0.0.1:9092",
    stream_timeout=10000,
    configuration=[
        "session.timeout.ms=7000",
        "max.poll.interval.ms=8000",
        "auto.offset.reset=earliest"
    ],
)

def decode_kafka_test_item(raw_message, raw_key):
  message = tf.io.decode_csv(raw_message, [[0.0] for i in range(NUM_COLUMNS)])
  key = tf.strings.to_number(raw_key)
  return (message, key)

test_ds = test_ds.map(decode_kafka_test_item)
test_ds = test_ds.batch(BATCH_SIZE)

Though this class can be used for training purposes, there are caveats which need to be addressed. Once all the messages are read from kafka and the latest offsets are committed using the streaming.KafkaGroupIODataset, the consumer doesn't restart reading the messages from the beginning. Thus, while training, it is possible only to train for a single epoch with the data continuously flowing in. This kind of a functionality has limited use cases during the training phase wherein, once a datapoint has been consumed by the model it is no longer required and can be discarded.

However, this functionality shines when it comes to robust inference with exactly-once semantics.

evaluate the performance on the test data

res = model.evaluate(test_ds)
print("test loss, test acc:", res)
625/625 [==============================] - 13s 21ms/step - loss: 0.4367 - accuracy: 0.7980
test loss, test acc: [0.4366855025291443, 0.7980499863624573]

Since the inference is based on 'exactly-once' semantics, the evaluation on the test set can be run only once. In order to run the inference again on the test data, a new consumer group should be used.

Track the offset lag of the testcg consumer group

./kafka_2.13-2.7.0/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group testcg
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                  HOST            CLIENT-ID
testcg          susy-test       0          21592           21592           0               rdkafka-66bc39ef-397c-4190-a17c-69ab519d4f53 /10.142.0.106   rdkafka
testcg          susy-test       1          18408           18408           0               rdkafka-66bc39ef-397c-4190-a17c-69ab519d4f53 /10.142.0.106   rdkafka

Once the current-offset matches the log-end-offset for all the partitions, it indicates that the consumer(s) have completed fetching all the messages from the kafka topic.

Online learning

The online machine learning paradigm is a bit different from the traditional/conventional way of training machine learning models. In the former case, the model continues to incrementally learn/update it's parameters as soon as the new data points are available and this process is expected to continue indefinitely. This is unlike the latter approaches where the dataset is fixed and the model iterates over it n number of times. In online learning, the data once consumed by the model may not be available for training again.

By utilizing the streaming.KafkaBatchIODataset, it is now possible to train the models in this fashion. Let's continue to use our SUSY dataset for demonstrating this functionality.

The tfio training dataset for online learning

The streaming.KafkaBatchIODataset is similar to the streaming.KafkaGroupIODataset in it's API. Additionally, it is recommended to utilize the stream_timeout parameter to configure the duration for which the dataset will block for new messages before timing out. In the instance below, the dataset is configured with a stream_timeout of 10000 milliseconds. This implies that, after all the messages from the topic have been consumed, the dataset will wait for an additional 10 seconds before timing out and disconnecting from the kafka cluster. If new messages are streamed into the topic before timing out, the data consumption and model training resumes for those newly consumed data points. To block indefinitely, set it to -1.

online_train_ds = tfio.experimental.streaming.KafkaBatchIODataset(
    topics=["susy-train"],
    group_id="cgonline",
    servers="127.0.0.1:9092",
    stream_timeout=10000, # in milliseconds, to block indefinitely, set it to -1.
    configuration=[
        "session.timeout.ms=7000",
        "max.poll.interval.ms=8000",
        "auto.offset.reset=earliest"
    ],
)

Every item that the online_train_ds generates is a tf.data.Dataset in itself. Thus, all the standard transformations can be applied as usual.

def decode_kafka_online_item(raw_message, raw_key):
  message = tf.io.decode_csv(raw_message, [[0.0] for i in range(NUM_COLUMNS)])
  key = tf.strings.to_number(raw_key)
  return (message, key)

for mini_ds in online_train_ds:
  mini_ds = mini_ds.shuffle(buffer_size=32)
  mini_ds = mini_ds.map(decode_kafka_online_item)
  mini_ds = mini_ds.batch(32)
  model.fit(mini_ds, epochs=3)
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4279 - accuracy: 0.8115
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4130 - accuracy: 0.8213
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.3996 - accuracy: 0.8242
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4646 - accuracy: 0.7920
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4415 - accuracy: 0.8047
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4296 - accuracy: 0.8018
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4410 - accuracy: 0.7969
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4223 - accuracy: 0.8174
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4187 - accuracy: 0.8096
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4679 - accuracy: 0.7734
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4292 - accuracy: 0.7979
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4411 - accuracy: 0.7959
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4750 - accuracy: 0.7695
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4460 - accuracy: 0.7959
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4345 - accuracy: 0.8008
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4282 - accuracy: 0.8154
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4086 - accuracy: 0.8184
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.3942 - accuracy: 0.8291
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4265 - accuracy: 0.7852
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4233 - accuracy: 0.7988
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4037 - accuracy: 0.8164
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4463 - accuracy: 0.7988
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4064 - accuracy: 0.8164
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4093 - accuracy: 0.8115
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4652 - accuracy: 0.7930
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4508 - accuracy: 0.7930
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4296 - accuracy: 0.8018
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4671 - accuracy: 0.7812
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4519 - accuracy: 0.7910
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4431 - accuracy: 0.7930
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4905 - accuracy: 0.7637
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4691 - accuracy: 0.7764
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4518 - accuracy: 0.7803
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4379 - accuracy: 0.7930
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4314 - accuracy: 0.8057
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4232 - accuracy: 0.7988
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4892 - accuracy: 0.7539
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4660 - accuracy: 0.7773
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4572 - accuracy: 0.7842
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4474 - accuracy: 0.7871
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4435 - accuracy: 0.7920
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4247 - accuracy: 0.8047
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4306 - accuracy: 0.8115
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4219 - accuracy: 0.8164
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4017 - accuracy: 0.8223
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4662 - accuracy: 0.7881
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4563 - accuracy: 0.7988
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4445 - accuracy: 0.7930
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4437 - accuracy: 0.7910
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4282 - accuracy: 0.7988
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4203 - accuracy: 0.8057
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4227 - accuracy: 0.7998
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4005 - accuracy: 0.8223
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.3905 - accuracy: 0.8184
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4617 - accuracy: 0.7871
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4414 - accuracy: 0.7891
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4320 - accuracy: 0.8018
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4493 - accuracy: 0.7900
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4418 - accuracy: 0.8037
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4265 - accuracy: 0.8096
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4740 - accuracy: 0.7783
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4498 - accuracy: 0.7812
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4473 - accuracy: 0.7910
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4439 - accuracy: 0.7900
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4217 - accuracy: 0.8047
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4126 - accuracy: 0.8135
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4417 - accuracy: 0.7998
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4120 - accuracy: 0.8105
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.3967 - accuracy: 0.8213
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4657 - accuracy: 0.7822
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4437 - accuracy: 0.7998
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4344 - accuracy: 0.8057
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4385 - accuracy: 0.8018
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4314 - accuracy: 0.8096
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4186 - accuracy: 0.8086
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4918 - accuracy: 0.7725
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4719 - accuracy: 0.7725
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4586 - accuracy: 0.7920
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4989 - accuracy: 0.7607
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4835 - accuracy: 0.7676
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4796 - accuracy: 0.7695
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4645 - accuracy: 0.7754
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4505 - accuracy: 0.7812
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4420 - accuracy: 0.7852
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4219 - accuracy: 0.8047
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4029 - accuracy: 0.8135
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.3870 - accuracy: 0.8252
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4627 - accuracy: 0.7871
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4313 - accuracy: 0.8066
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4324 - accuracy: 0.8066
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4532 - accuracy: 0.7959
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4370 - accuracy: 0.7871
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4301 - accuracy: 0.8027
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4319 - accuracy: 0.7910
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4268 - accuracy: 0.7979
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4096 - accuracy: 0.8154
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4738 - accuracy: 0.7734
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4411 - accuracy: 0.7842
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4410 - accuracy: 0.7754
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4437 - accuracy: 0.7881
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4273 - accuracy: 0.8047
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4030 - accuracy: 0.8193
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4604 - accuracy: 0.7725
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4432 - accuracy: 0.7744
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4299 - accuracy: 0.7969
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4276 - accuracy: 0.8037
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4138 - accuracy: 0.8076
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4009 - accuracy: 0.8105
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4273 - accuracy: 0.8086
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4049 - accuracy: 0.8193
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.3974 - accuracy: 0.8252
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4755 - accuracy: 0.7705
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4454 - accuracy: 0.7832
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4358 - accuracy: 0.7939
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4691 - accuracy: 0.7686
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4572 - accuracy: 0.7793
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4484 - accuracy: 0.7979
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4292 - accuracy: 0.8066
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4220 - accuracy: 0.8145
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4183 - accuracy: 0.8115
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4420 - accuracy: 0.7930
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4384 - accuracy: 0.7939
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4354 - accuracy: 0.7881
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4731 - accuracy: 0.7666
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4544 - accuracy: 0.7764
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4451 - accuracy: 0.7812
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4211 - accuracy: 0.8047
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4097 - accuracy: 0.8047
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4013 - accuracy: 0.8164
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4625 - accuracy: 0.7832
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4405 - accuracy: 0.7998
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4224 - accuracy: 0.8066
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4404 - accuracy: 0.7881
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4227 - accuracy: 0.7959
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4079 - accuracy: 0.8066
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4570 - accuracy: 0.7910
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4355 - accuracy: 0.8135
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4281 - accuracy: 0.7998
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4721 - accuracy: 0.7891
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4606 - accuracy: 0.7969
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4401 - accuracy: 0.8008
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4666 - accuracy: 0.7842
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4574 - accuracy: 0.7871
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4460 - accuracy: 0.7920
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4538 - accuracy: 0.7969
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4420 - accuracy: 0.8047
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4189 - accuracy: 0.8086
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4766 - accuracy: 0.7627
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4589 - accuracy: 0.7725
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4550 - accuracy: 0.7725
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4338 - accuracy: 0.8018
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4217 - accuracy: 0.8057
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4044 - accuracy: 0.8154
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4523 - accuracy: 0.7881
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4451 - accuracy: 0.7920
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4208 - accuracy: 0.8018
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4711 - accuracy: 0.7666
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4518 - accuracy: 0.7852
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4293 - accuracy: 0.7930
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4786 - accuracy: 0.7783
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4586 - accuracy: 0.7842
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4426 - accuracy: 0.7881
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4392 - accuracy: 0.8115
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4181 - accuracy: 0.8252
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4134 - accuracy: 0.8223
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4380 - accuracy: 0.7998
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4271 - accuracy: 0.8018
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4118 - accuracy: 0.8105
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4469 - accuracy: 0.7939
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4358 - accuracy: 0.7959
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4253 - accuracy: 0.8018
Epoch 1/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4574 - accuracy: 0.8018
Epoch 2/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4531 - accuracy: 0.8018
Epoch 3/3
32/32 [==============================] - 0s 3ms/step - loss: 0.4392 - accuracy: 0.7988
Epoch 1/3
19/19 [==============================] - 0s 3ms/step - loss: 0.4244 - accuracy: 0.8191
Epoch 2/3
19/19 [==============================] - 0s 3ms/step - loss: 0.4049 - accuracy: 0.8339
Epoch 3/3
19/19 [==============================] - 0s 3ms/step - loss: 0.3980 - accuracy: 0.8224

The incrementally trained model can be saved in a periodic fashion (based on use-cases) and can be utilized to infer on the test data in either online or offline modes.

References: