Streaming_Data.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyNvi6Ae7GgDYusqdiFcgEBo",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/SharmaNatasha/Machine-Learning-using-Python/blob/master/Streaming_Data.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "RbOywBQRxyBz"
      },
      "outputs": [],
      "source": [
        "!pip uninstall tensorflow\n",
        "!pip uninstall tensorflow_io\n",
        "!pip install tensorflow==2.7.1\n",
        "!pip install tensorflow_io==0.23.1\n",
        "!pip install kafka-python"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import os\n",
        "from datetime import datetime\n",
        "import time\n",
        "import threading\n",
        "import json\n",
        "from kafka import KafkaProducer\n",
        "from kafka.errors import KafkaError\n",
        "# SKLearn libraries\n",
        "from sklearn.preprocessing import LabelEncoder\n",
        "from sklearn.model_selection import train_test_split\n",
        "import pandas as pd\n",
        "import tensorflow as tf\n",
        "import tensorflow_io as tfio"
      ],
      "metadata": {
        "id": "xfrczVz6xzbA"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!curl -sSOL https://downloads.apache.org/kafka/3.3.2/kafka_2.13-3.3.2.tgz\n",
        "!tar -xzf kafka_2.13-3.3.2.tgz"
      ],
      "metadata": {
        "id": "NVwEHpsqIGAv"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!./kafka_2.13-3.3.2/bin/zookeeper-server-start.sh -daemon ./kafka_2.13-3.3.2/config/zookeeper.properties\n",
        "!./kafka_2.13-3.3.2/bin/kafka-server-start.sh -daemon ./kafka_2.13-3.3.2/config/server.properties\n",
        "!echo \"Waiting for 10 secs until kafka and zookeeper services are up and running\"\n",
        "!sleep 10"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "RG9iuB_8Iub0",
        "outputId": "640f8b26-34c6-4b3f-bba6-a7a02dfe4805"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Waiting for 10 secs until kafka and zookeeper services are up and running\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!ps -ef | grep kafka"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "mI8LdRAbKOXp",
        "outputId": "c7cb953f-e753-47ab-994a-09b4848bede8"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "root       11334       1 13 10:46 ?        00:00:02 java -Xmx512M -Xms512M -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true -Xlog:gc*:file=/content/kafka_2.13-3.3.2/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=/content/kafka_2.13-3.3.2/bin/../logs -Dlog4j.configuration=file:./kafka_2.13-3.3.2/bin/../config/log4j.properties -cp /content/kafka_2.13-3.3.2/bin/../libs/activation-1.1.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/aopalliance-repackaged-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/argparse4j-0.7.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/audience-annotations-0.5.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/commons-cli-1.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/commons-lang3-3.12.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/commons-lang3-3.8.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-api-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-basic-auth-extension-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-json-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-mirror-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-mirror-client-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-runtime-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-transforms-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/hk2-api-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/hk2-locator-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/hk2-utils-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-annotations-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-core-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-databind-2.13.4.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-dataformat-csv-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-datatype-jdk8-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-jaxrs-base-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-jaxrs-json-provider-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-module-jaxb-annotations-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-module-scala_2.13-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.activation-api-1.2.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.annotation-api-1.3.5.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.inject-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.validation-api-2.0.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.ws.rs-api-2.1.6.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.xml.bind-api-2.3.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/javassist-3.27.0-GA.jar:/content/kafka_2.13-3.3.2/bin/../libs/javax.servlet-api-3.1.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/javax.ws.rs-api-2.1.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/jaxb-api-2.3.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-client-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-common-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-container-servlet-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-container-servlet-core-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-hk2-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-server-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-client-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-continuation-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-http-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-io-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-security-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-server-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-servlet-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-servlets-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-util-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-util-ajax-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jline-3.21.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/jopt-simple-5.0.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jose4j-0.7.9.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka_2.13-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-clients-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-log4j-appender-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-metadata-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-raft-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-server-common-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-shell-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-storage-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-storage-api-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-examples-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-scala_2.13-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-test-utils-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-tools-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/lz4-java-1.8.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/maven-artifact-3.8.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/metrics-core-2.2.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/metrics-core-4.1.12.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-buffer-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-codec-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-common-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-handler-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-resolver-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-classes-epoll-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-native-epoll-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-native-unix-common-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/osgi-resource-locator-1.0.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/paranamer-2.8.jar:/content/kafka_2.13-3.3.2/bin/../libs/plexus-utils-3.3.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/reflections-0.9.12.jar:/content/kafka_2.13-3.3.2/bin/../libs/reload4j-1.2.19.jar:/content/kafka_2.13-3.3.2/bin/../libs/rocksdbjni-7.1.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-collection-compat_2.13-2.6.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-java8-compat_2.13-1.0.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-library-2.13.8.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-logging_2.13-3.9.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-reflect-2.13.8.jar:/content/kafka_2.13-3.3.2/bin/../libs/slf4j-api-1.7.36.jar:/content/kafka_2.13-3.3.2/bin/../libs/slf4j-reload4j-1.7.36.jar:/content/kafka_2.13-3.3.2/bin/../libs/snappy-java-1.1.8.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/swagger-annotations-2.2.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/trogdor-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/zookeeper-3.6.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/zookeeper-jute-3.6.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/zstd-jni-1.5.2-1.jar org.apache.zookeeper.server.quorum.QuorumPeerMain ./kafka_2.13-3.3.2/config/zookeeper.properties\n",
            "root       11704       1 51 10:46 ?        00:00:07 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true -Xlog:gc*:file=/content/kafka_2.13-3.3.2/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=/content/kafka_2.13-3.3.2/bin/../logs -Dlog4j.configuration=file:./kafka_2.13-3.3.2/bin/../config/log4j.properties -cp /content/kafka_2.13-3.3.2/bin/../libs/activation-1.1.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/aopalliance-repackaged-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/argparse4j-0.7.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/audience-annotations-0.5.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/commons-cli-1.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/commons-lang3-3.12.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/commons-lang3-3.8.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-api-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-basic-auth-extension-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-json-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-mirror-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-mirror-client-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-runtime-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/connect-transforms-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/hk2-api-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/hk2-locator-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/hk2-utils-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-annotations-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-core-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-databind-2.13.4.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-dataformat-csv-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-datatype-jdk8-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-jaxrs-base-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-jaxrs-json-provider-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-module-jaxb-annotations-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jackson-module-scala_2.13-2.13.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.activation-api-1.2.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.annotation-api-1.3.5.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.inject-2.6.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.validation-api-2.0.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.ws.rs-api-2.1.6.jar:/content/kafka_2.13-3.3.2/bin/../libs/jakarta.xml.bind-api-2.3.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/javassist-3.27.0-GA.jar:/content/kafka_2.13-3.3.2/bin/../libs/javax.servlet-api-3.1.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/javax.ws.rs-api-2.1.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/jaxb-api-2.3.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-client-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-common-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-container-servlet-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-container-servlet-core-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-hk2-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jersey-server-2.34.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-client-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-continuation-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-http-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-io-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-security-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-server-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-servlet-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-servlets-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-util-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jetty-util-ajax-9.4.48.v20220622.jar:/content/kafka_2.13-3.3.2/bin/../libs/jline-3.21.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/jopt-simple-5.0.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/jose4j-0.7.9.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka_2.13-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-clients-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-log4j-appender-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-metadata-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-raft-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-server-common-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-shell-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-storage-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-storage-api-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-examples-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-scala_2.13-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-streams-test-utils-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/kafka-tools-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/lz4-java-1.8.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/maven-artifact-3.8.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/metrics-core-2.2.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/metrics-core-4.1.12.1.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-buffer-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-codec-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-common-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-handler-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-resolver-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-classes-epoll-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-native-epoll-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/netty-transport-native-unix-common-4.1.78.Final.jar:/content/kafka_2.13-3.3.2/bin/../libs/osgi-resource-locator-1.0.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/paranamer-2.8.jar:/content/kafka_2.13-3.3.2/bin/../libs/plexus-utils-3.3.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/reflections-0.9.12.jar:/content/kafka_2.13-3.3.2/bin/../libs/reload4j-1.2.19.jar:/content/kafka_2.13-3.3.2/bin/../libs/rocksdbjni-7.1.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-collection-compat_2.13-2.6.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-java8-compat_2.13-1.0.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-library-2.13.8.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-logging_2.13-3.9.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/scala-reflect-2.13.8.jar:/content/kafka_2.13-3.3.2/bin/../libs/slf4j-api-1.7.36.jar:/content/kafka_2.13-3.3.2/bin/../libs/slf4j-reload4j-1.7.36.jar:/content/kafka_2.13-3.3.2/bin/../libs/snappy-java-1.1.8.4.jar:/content/kafka_2.13-3.3.2/bin/../libs/swagger-annotations-2.2.0.jar:/content/kafka_2.13-3.3.2/bin/../libs/trogdor-3.3.2.jar:/content/kafka_2.13-3.3.2/bin/../libs/zookeeper-3.6.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/zookeeper-jute-3.6.3.jar:/content/kafka_2.13-3.3.2/bin/../libs/zstd-jni-1.5.2-1.jar kafka.Kafka ./kafka_2.13-3.3.2/config/server.properties\n",
            "root       11842     167  0 10:46 ?        00:00:00 /bin/bash -c ps -ef | grep kafka\n",
            "root       11844   11842  0 10:46 ?        00:00:00 grep kafka\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!./kafka_2.13-3.3.2/bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --replication-factor 1 --partitions 1 --topic cancer-train\n",
        "!./kafka_2.13-3.3.2/bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --replication-factor 1 --partitions 2 --topic cancer-test"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "VISghZXxKuVS",
        "outputId": "d94982f9-1dcb-4a65-fa11-0a74d7f1cc60"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Created topic cancer-train.\n",
            "Created topic cancer-test.\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "cancer_df = pd.read_csv('breast-cancer-wisconsin.data.csv')\n",
        "cancer_df.head()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 268
        },
        "id": "0VyKMSkbNRqz",
        "outputId": "b7910155-4e87-42e7-cd8f-dfde2a1c3870"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "   Sample code number  Clump Thickness  Uniformity of Cell Size  \\\n",
              "0             1000025                5                        1   \n",
              "1             1002945                5                        4   \n",
              "2             1015425                3                        1   \n",
              "3             1016277                6                        8   \n",
              "4             1017023                4                        1   \n",
              "\n",
              "   uniformity of Cell Shape  Marginal Adhesion  Single Epithelial Cell Size  \\\n",
              "0                         1                  1                            2   \n",
              "1                         4                  5                            7   \n",
              "2                         1                  1                            2   \n",
              "3                         8                  1                            3   \n",
              "4                         1                  3                            2   \n",
              "\n",
              "  Bare Nuclei  Bland Chromatin  Normal Nucleoli  Mitoses  Class  \n",
              "0           1                3                1        1      2  \n",
              "1          10                3                2        1      2  \n",
              "2           2                3                1        1      2  \n",
              "3           4                3                7        1      2  \n",
              "4           1                3                1        1      2  "
            ],
            "text/html": [
              "\n",
              "  <div id=\"df-41f94a08-7ed8-4a52-995d-037c24e52a2e\">\n",
              "    <div class=\"colab-df-container\">\n",
              "      <div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>Sample code number</th>\n",
              "      <th>Clump Thickness</th>\n",
              "      <th>Uniformity of Cell Size</th>\n",
              "      <th>uniformity of Cell Shape</th>\n",
              "      <th>Marginal Adhesion</th>\n",
              "      <th>Single Epithelial Cell Size</th>\n",
              "      <th>Bare Nuclei</th>\n",
              "      <th>Bland Chromatin</th>\n",
              "      <th>Normal Nucleoli</th>\n",
              "      <th>Mitoses</th>\n",
              "      <th>Class</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>1000025</td>\n",
              "      <td>5</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "      <td>1</td>\n",
              "      <td>3</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>1002945</td>\n",
              "      <td>5</td>\n",
              "      <td>4</td>\n",
              "      <td>4</td>\n",
              "      <td>5</td>\n",
              "      <td>7</td>\n",
              "      <td>10</td>\n",
              "      <td>3</td>\n",
              "      <td>2</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>1015425</td>\n",
              "      <td>3</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "      <td>2</td>\n",
              "      <td>3</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>1016277</td>\n",
              "      <td>6</td>\n",
              "      <td>8</td>\n",
              "      <td>8</td>\n",
              "      <td>1</td>\n",
              "      <td>3</td>\n",
              "      <td>4</td>\n",
              "      <td>3</td>\n",
              "      <td>7</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td>1017023</td>\n",
              "      <td>4</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>3</td>\n",
              "      <td>2</td>\n",
              "      <td>1</td>\n",
              "      <td>3</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>\n",
              "      <button class=\"colab-df-convert\" onclick=\"convertToInteractive('df-41f94a08-7ed8-4a52-995d-037c24e52a2e')\"\n",
              "              title=\"Convert this dataframe to an interactive table.\"\n",
              "              style=\"display:none;\">\n",
              "        \n",
              "  <svg xmlns=\"http://www.w3.org/2000/svg\" height=\"24px\"viewBox=\"0 0 24 24\"\n",
              "       width=\"24px\">\n",
              "    <path d=\"M0 0h24v24H0V0z\" fill=\"none\"/>\n",
              "    <path d=\"M18.56 5.44l.94 2.06.94-2.06 2.06-.94-2.06-.94-.94-2.06-.94 2.06-2.06.94zm-11 1L8.5 8.5l.94-2.06 2.06-.94-2.06-.94L8.5 2.5l-.94 2.06-2.06.94zm10 10l.94 2.06.94-2.06 2.06-.94-2.06-.94-.94-2.06-.94 2.06-2.06.94z\"/><path d=\"M17.41 7.96l-1.37-1.37c-.4-.4-.92-.59-1.43-.59-.52 0-1.04.2-1.43.59L10.3 9.45l-7.72 7.72c-.78.78-.78 2.05 0 2.83L4 21.41c.39.39.9.59 1.41.59.51 0 1.02-.2 1.41-.59l7.78-7.78 2.81-2.81c.8-.78.8-2.07 0-2.86zM5.41 20L4 18.59l7.72-7.72 1.47 1.35L5.41 20z\"/>\n",
              "  </svg>\n",
              "      </button>\n",
              "      \n",
              "  <style>\n",
              "    .colab-df-container {\n",
              "      display:flex;\n",
              "      flex-wrap:wrap;\n",
              "      gap: 12px;\n",
              "    }\n",
              "\n",
              "    .colab-df-convert {\n",
              "      background-color: #E8F0FE;\n",
              "      border: none;\n",
              "      border-radius: 50%;\n",
              "      cursor: pointer;\n",
              "      display: none;\n",
              "      fill: #1967D2;\n",
              "      height: 32px;\n",
              "      padding: 0 0 0 0;\n",
              "      width: 32px;\n",
              "    }\n",
              "\n",
              "    .colab-df-convert:hover {\n",
              "      background-color: #E2EBFA;\n",
              "      box-shadow: 0px 1px 2px rgba(60, 64, 67, 0.3), 0px 1px 3px 1px rgba(60, 64, 67, 0.15);\n",
              "      fill: #174EA6;\n",
              "    }\n",
              "\n",
              "    [theme=dark] .colab-df-convert {\n",
              "      background-color: #3B4455;\n",
              "      fill: #D2E3FC;\n",
              "    }\n",
              "\n",
              "    [theme=dark] .colab-df-convert:hover {\n",
              "      background-color: #434B5C;\n",
              "      box-shadow: 0px 1px 3px 1px rgba(0, 0, 0, 0.15);\n",
              "      filter: drop-shadow(0px 1px 2px rgba(0, 0, 0, 0.3));\n",
              "      fill: #FFFFFF;\n",
              "    }\n",
              "  </style>\n",
              "\n",
              "      <script>\n",
              "        const buttonEl =\n",
              "          document.querySelector('#df-41f94a08-7ed8-4a52-995d-037c24e52a2e button.colab-df-convert');\n",
              "        buttonEl.style.display =\n",
              "          google.colab.kernel.accessAllowed ? 'block' : 'none';\n",
              "\n",
              "        async function convertToInteractive(key) {\n",
              "          const element = document.querySelector('#df-41f94a08-7ed8-4a52-995d-037c24e52a2e');\n",
              "          const dataTable =\n",
              "            await google.colab.kernel.invokeFunction('convertToInteractive',\n",
              "                                                     [key], {});\n",
              "          if (!dataTable) return;\n",
              "\n",
              "          const docLinkHtml = 'Like what you see? Visit the ' +\n",
              "            '<a target=\"_blank\" href=https://colab.research.google.com/notebooks/data_table.ipynb>data table notebook</a>'\n",
              "            + ' to learn more about interactive tables.';\n",
              "          element.innerHTML = '';\n",
              "          dataTable['output_type'] = 'display_data';\n",
              "          await google.colab.output.renderOutput(dataTable, element);\n",
              "          const docLink = document.createElement('div');\n",
              "          docLink.innerHTML = docLinkHtml;\n",
              "          element.appendChild(docLink);\n",
              "        }\n",
              "      </script>\n",
              "    </div>\n",
              "  </div>\n",
              "  "
            ]
          },
          "metadata": {},
          "execution_count": 9
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "cancer_df.dtypes"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "S_V5L94xFHCa",
        "outputId": "76ce285f-6427-4d33-9b61-316c02b1d9eb"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "Sample code number              int64\n",
              "Clump Thickness                 int64\n",
              "Uniformity of Cell Size         int64\n",
              "uniformity of Cell Shape        int64\n",
              "Marginal Adhesion               int64\n",
              "Single Epithelial Cell Size     int64\n",
              "Bare Nuclei                    object\n",
              "Bland Chromatin                 int64\n",
              "Normal Nucleoli                 int64\n",
              "Mitoses                         int64\n",
              "Class                           int64\n",
              "dtype: object"
            ]
          },
          "metadata": {},
          "execution_count": 129
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Number of datapoints and columns\n",
        "len(cancer_df), len(cancer_df.columns)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Xo0tD3FQNUQW",
        "outputId": "d6d93559-9f70-4194-b3e6-607592a7b959"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "(699, 11)"
            ]
          },
          "metadata": {},
          "execution_count": 130
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "cancer_df['Class'] = cancer_df['Class'].replace(2,0)\n",
        "cancer_df['Class'] = cancer_df['Class'].replace(4,1)"
      ],
      "metadata": {
        "id": "LmNHLz6kE3Q1"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "len(cancer_df[cancer_df[\"Class\"]==0]), len(cancer_df[cancer_df[\"Class\"]==1])"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "k24VSPYmRSCV",
        "outputId": "d045a896-fd5a-4f7d-8797-011cf110dc54"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "(458, 241)"
            ]
          },
          "metadata": {},
          "execution_count": 132
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "cancer_df = cancer_df.drop(['Bare Nuclei'], axis=1)\n",
        "cancer_df.dtypes"
      ],
      "metadata": {
        "id": "vZ2FtA8RRdAd",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "03fb513a-8f3f-48f6-e848-a46e00b7a0f6"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "Sample code number             int64\n",
              "Clump Thickness                int64\n",
              "Uniformity of Cell Size        int64\n",
              "uniformity of Cell Shape       int64\n",
              "Marginal Adhesion              int64\n",
              "Single Epithelial Cell Size    int64\n",
              "Bland Chromatin                int64\n",
              "Normal Nucleoli                int64\n",
              "Mitoses                        int64\n",
              "Class                          int64\n",
              "dtype: object"
            ]
          },
          "metadata": {},
          "execution_count": 11
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "cancer_df.isnull().sum()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "I6XptC9eY3GQ",
        "outputId": "c3562d90-7e66-4c76-c6f9-1e79664142f5"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "Sample code number             0\n",
              "Clump Thickness                0\n",
              "Uniformity of Cell Size        0\n",
              "uniformity of Cell Shape       0\n",
              "Marginal Adhesion              0\n",
              "Single Epithelial Cell Size    0\n",
              "Bland Chromatin                0\n",
              "Normal Nucleoli                0\n",
              "Mitoses                        0\n",
              "Class                          0\n",
              "dtype: int64"
            ]
          },
          "metadata": {},
          "execution_count": 135
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "train_df, test_df = train_test_split(cancer_df, test_size=0.4, shuffle=True)\n",
        "print(\"Number of training samples: \",len(train_df))\n",
        "print(\"Number of testing sample: \",len(test_df))\n",
        "\n",
        "x_train_df = train_df.drop([\"Class\"], axis=1)\n",
        "y_train_df = train_df[\"Class\"]\n",
        "\n",
        "x_test_df = test_df.drop([\"Class\"], axis=1)\n",
        "y_test_df = test_df[\"Class\"]\n",
        "\n",
        "len(x_train_df.columns)"
      ],
      "metadata": {
        "id": "ul6yuRuhnqVt",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "35f9c222-5d99-45fb-f8f2-8281f3651b75"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Number of training samples:  419\n",
            "Number of testing sample:  280\n"
          ]
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "9"
            ]
          },
          "metadata": {},
          "execution_count": 12
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "x_train_df.to_csv(index=False).split(\"\\n\")[1:]"
      ],
      "metadata": {
        "id": "qELkxMhbbWd8"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# The labels are set as the kafka message keys so as to store data\n",
        "# in multiple-partitions. Thus, enabling efficient data retrieval\n",
        "# using the consumer groups.\n",
        "x_train = list(filter(None, x_train_df.to_csv(index=False).split(\"\\n\")[1:]))\n",
        "y_train = list(filter(None, y_train_df.to_csv(index=False).split(\"\\n\")[1:]))\n",
        "\n",
        "x_test = list(filter(None, x_test_df.to_csv(index=False).split(\"\\n\")[1:]))\n",
        "y_test = list(filter(None, y_test_df.to_csv(index=False).split(\"\\n\")[1:]))"
      ],
      "metadata": {
        "id": "CW2eov2SbNWw"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "NUM_COLUMNS = len(x_train_df.columns)\n",
        "print(NUM_COLUMNS)\n",
        "len(x_train), len(y_train), len(x_test), len(y_test)"
      ],
      "metadata": {
        "id": "qkI-xTW-NY5s",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "32fa4055-972a-4116-dc35-69b66af0b53d"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "9\n"
          ]
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "(419, 419, 280, 280)"
            ]
          },
          "metadata": {},
          "execution_count": 17
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "x_train_df.head(1)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 159
        },
        "id": "WOv72eywVbMX",
        "outputId": "5a066261-2fe0-439e-f1b5-d01299b5a7f3"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "   Sample code number  Clump Thickness  Uniformity of Cell Size  \\\n",
              "6             1018099                1                        1   \n",
              "\n",
              "   uniformity of Cell Shape  Marginal Adhesion  Single Epithelial Cell Size  \\\n",
              "6                         1                  1                            2   \n",
              "\n",
              "   Bland Chromatin  Normal Nucleoli  Mitoses  \n",
              "6                3                1        1  "
            ],
            "text/html": [
              "\n",
              "  <div id=\"df-c4c5706c-7e0d-45c5-8169-eca4c545f23e\">\n",
              "    <div class=\"colab-df-container\">\n",
              "      <div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>Sample code number</th>\n",
              "      <th>Clump Thickness</th>\n",
              "      <th>Uniformity of Cell Size</th>\n",
              "      <th>uniformity of Cell Shape</th>\n",
              "      <th>Marginal Adhesion</th>\n",
              "      <th>Single Epithelial Cell Size</th>\n",
              "      <th>Bland Chromatin</th>\n",
              "      <th>Normal Nucleoli</th>\n",
              "      <th>Mitoses</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>6</th>\n",
              "      <td>1018099</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "      <td>2</td>\n",
              "      <td>3</td>\n",
              "      <td>1</td>\n",
              "      <td>1</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>\n",
              "      <button class=\"colab-df-convert\" onclick=\"convertToInteractive('df-c4c5706c-7e0d-45c5-8169-eca4c545f23e')\"\n",
              "              title=\"Convert this dataframe to an interactive table.\"\n",
              "              style=\"display:none;\">\n",
              "        \n",
              "  <svg xmlns=\"http://www.w3.org/2000/svg\" height=\"24px\"viewBox=\"0 0 24 24\"\n",
              "       width=\"24px\">\n",
              "    <path d=\"M0 0h24v24H0V0z\" fill=\"none\"/>\n",
              "    <path d=\"M18.56 5.44l.94 2.06.94-2.06 2.06-.94-2.06-.94-.94-2.06-.94 2.06-2.06.94zm-11 1L8.5 8.5l.94-2.06 2.06-.94-2.06-.94L8.5 2.5l-.94 2.06-2.06.94zm10 10l.94 2.06.94-2.06 2.06-.94-2.06-.94-.94-2.06-.94 2.06-2.06.94z\"/><path d=\"M17.41 7.96l-1.37-1.37c-.4-.4-.92-.59-1.43-.59-.52 0-1.04.2-1.43.59L10.3 9.45l-7.72 7.72c-.78.78-.78 2.05 0 2.83L4 21.41c.39.39.9.59 1.41.59.51 0 1.02-.2 1.41-.59l7.78-7.78 2.81-2.81c.8-.78.8-2.07 0-2.86zM5.41 20L4 18.59l7.72-7.72 1.47 1.35L5.41 20z\"/>\n",
              "  </svg>\n",
              "      </button>\n",
              "      \n",
              "  <style>\n",
              "    .colab-df-container {\n",
              "      display:flex;\n",
              "      flex-wrap:wrap;\n",
              "      gap: 12px;\n",
              "    }\n",
              "\n",
              "    .colab-df-convert {\n",
              "      background-color: #E8F0FE;\n",
              "      border: none;\n",
              "      border-radius: 50%;\n",
              "      cursor: pointer;\n",
              "      display: none;\n",
              "      fill: #1967D2;\n",
              "      height: 32px;\n",
              "      padding: 0 0 0 0;\n",
              "      width: 32px;\n",
              "    }\n",
              "\n",
              "    .colab-df-convert:hover {\n",
              "      background-color: #E2EBFA;\n",
              "      box-shadow: 0px 1px 2px rgba(60, 64, 67, 0.3), 0px 1px 3px 1px rgba(60, 64, 67, 0.15);\n",
              "      fill: #174EA6;\n",
              "    }\n",
              "\n",
              "    [theme=dark] .colab-df-convert {\n",
              "      background-color: #3B4455;\n",
              "      fill: #D2E3FC;\n",
              "    }\n",
              "\n",
              "    [theme=dark] .colab-df-convert:hover {\n",
              "      background-color: #434B5C;\n",
              "      box-shadow: 0px 1px 3px 1px rgba(0, 0, 0, 0.15);\n",
              "      filter: drop-shadow(0px 1px 2px rgba(0, 0, 0, 0.3));\n",
              "      fill: #FFFFFF;\n",
              "    }\n",
              "  </style>\n",
              "\n",
              "      <script>\n",
              "        const buttonEl =\n",
              "          document.querySelector('#df-c4c5706c-7e0d-45c5-8169-eca4c545f23e button.colab-df-convert');\n",
              "        buttonEl.style.display =\n",
              "          google.colab.kernel.accessAllowed ? 'block' : 'none';\n",
              "\n",
              "        async function convertToInteractive(key) {\n",
              "          const element = document.querySelector('#df-c4c5706c-7e0d-45c5-8169-eca4c545f23e');\n",
              "          const dataTable =\n",
              "            await google.colab.kernel.invokeFunction('convertToInteractive',\n",
              "                                                     [key], {});\n",
              "          if (!dataTable) return;\n",
              "\n",
              "          const docLinkHtml = 'Like what you see? Visit the ' +\n",
              "            '<a target=\"_blank\" href=https://colab.research.google.com/notebooks/data_table.ipynb>data table notebook</a>'\n",
              "            + ' to learn more about interactive tables.';\n",
              "          element.innerHTML = '';\n",
              "          dataTable['output_type'] = 'display_data';\n",
              "          await google.colab.output.renderOutput(dataTable, element);\n",
              "          const docLink = document.createElement('div');\n",
              "          docLink.innerHTML = docLinkHtml;\n",
              "          element.appendChild(docLink);\n",
              "        }\n",
              "      </script>\n",
              "    </div>\n",
              "  </div>\n",
              "  "
            ]
          },
          "metadata": {},
          "execution_count": 144
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "def error_callback(exc):\n",
        "      raise Exception('Error while sendig data to kafka: {0}'.format(str(exc)))\n",
        "\n",
        "def write_to_kafka(topic_name, items):\n",
        "      count=0\n",
        "      producer = KafkaProducer(bootstrap_servers=['127.0.0.1:9092'])\n",
        "      for message, key in items:\n",
        "        producer.send(topic_name,\n",
        "                      key=key.encode('utf-8'),\n",
        "                      value=message.encode('utf-8')).add_errback(error_callback)\n",
        "        count+=1\n",
        "      producer.flush()\n",
        "      print(\"Wrote {0} messages into topic: {1}\".format(count, topic_name))\n",
        "\n",
        "write_to_kafka(\"cancer-train\", zip(x_train, y_train))\n",
        "write_to_kafka(\"cancer-test\", zip(x_test, y_test))"
      ],
      "metadata": {
        "id": "JkfOlt6JOciP",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "80da2baa-7c84-4650-bad7-0067fbc3d143"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Wrote 419 messages into topic: cancer-train\n",
            "Wrote 280 messages into topic: cancer-test\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "def decode_kafka_item(item):\n",
        "  message = tf.io.decode_csv(item.message, [[0.0] for i in range(NUM_COLUMNS)])\n",
        "  key = tf.strings.to_number(item.key)\n",
        "  return (message, key)\n",
        "\n",
        "BATCH_SIZE=64\n",
        "SHUFFLE_BUFFER_SIZE=64\n",
        "\n",
        "train_ds = tfio.IODataset.from_kafka('cancer-train1', partition=0, offset=0)\n",
        "train_ds = train_ds.shuffle(buffer_size=SHUFFLE_BUFFER_SIZE)\n",
        "train_ds = train_ds.map(decode_kafka_item)\n",
        "train_ds = train_ds.batch(BATCH_SIZE)"
      ],
      "metadata": {
        "id": "3bR6FOYdP-Pt"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Set the parameters\n",
        "\n",
        "OPTIMIZER=\"adam\"\n",
        "LOSS=tf.keras.losses.BinaryCrossentropy(from_logits=True)\n",
        "METRICS=['accuracy']\n",
        "EPOCHS=10"
      ],
      "metadata": {
        "id": "F1rJYVhQQN-j"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# design/build the model\n",
        "model = tf.keras.Sequential([\n",
        "  tf.keras.layers.Input(shape=(9,)),\n",
        "  tf.keras.layers.Dense(128, activation='relu'),\n",
        "  tf.keras.layers.Dropout(0.2),\n",
        "  tf.keras.layers.Dense(256, activation='relu'),\n",
        "  tf.keras.layers.Dropout(0.4),\n",
        "  tf.keras.layers.Dense(128, activation='relu'),\n",
        "  tf.keras.layers.Dropout(0.4),\n",
        "  tf.keras.layers.Dense(1, activation='sigmoid')\n",
        "])\n",
        "\n",
        "print(model.summary())"
      ],
      "metadata": {
        "id": "fV4nQdJWddkx",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "0bdfe7c8-d723-403b-9d10-b775841b126f"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Model: \"sequential\"\n",
            "_________________________________________________________________\n",
            " Layer (type)                Output Shape              Param #   \n",
            "=================================================================\n",
            " dense (Dense)               (None, 128)               1280      \n",
            "                                                                 \n",
            " dropout (Dropout)           (None, 128)               0         \n",
            "                                                                 \n",
            " dense_1 (Dense)             (None, 256)               33024     \n",
            "                                                                 \n",
            " dropout_1 (Dropout)         (None, 256)               0         \n",
            "                                                                 \n",
            " dense_2 (Dense)             (None, 128)               32896     \n",
            "                                                                 \n",
            " dropout_2 (Dropout)         (None, 128)               0         \n",
            "                                                                 \n",
            " dense_3 (Dense)             (None, 1)                 129       \n",
            "                                                                 \n",
            "=================================================================\n",
            "Total params: 67,329\n",
            "Trainable params: 67,329\n",
            "Non-trainable params: 0\n",
            "_________________________________________________________________\n",
            "None\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# compile the model\n",
        "model.compile(optimizer=OPTIMIZER, loss=LOSS, metrics=METRICS)"
      ],
      "metadata": {
        "id": "TFkJM5yBdnRO"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "model.fit(train_ds, epochs=EPOCHS)"
      ],
      "metadata": {
        "id": "UgrDQ44Ydt93",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "b8f0ce3e-c9b9-465a-cfe9-d2176847a56a"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Epoch 1/10\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/usr/local/lib/python3.8/dist-packages/tensorflow/python/util/dispatch.py:1096: UserWarning: \"`binary_crossentropy` received `from_logits=True`, but the `output` argument was produced by a sigmoid or softmax activation and thus does not represent logits. Was this intended?\"\n",
            "  return dispatch_target(*args, **kwargs)\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "7/7 [==============================] - 1s 20ms/step - loss: 32113.9238 - accuracy: 0.5513\n",
            "Epoch 2/10\n",
            "7/7 [==============================] - 0s 7ms/step - loss: 24886.2812 - accuracy: 0.5298\n",
            "Epoch 3/10\n",
            "7/7 [==============================] - 0s 7ms/step - loss: 18030.4375 - accuracy: 0.5537\n",
            "Epoch 4/10\n",
            "7/7 [==============================] - 0s 5ms/step - loss: 17029.5898 - accuracy: 0.5394\n",
            "Epoch 5/10\n",
            "7/7 [==============================] - 0s 6ms/step - loss: 11502.6982 - accuracy: 0.5489\n",
            "Epoch 6/10\n",
            "7/7 [==============================] - 0s 6ms/step - loss: 8582.7080 - accuracy: 0.5680\n",
            "Epoch 7/10\n",
            "7/7 [==============================] - 0s 6ms/step - loss: 9097.5449 - accuracy: 0.5489\n",
            "Epoch 8/10\n",
            "7/7 [==============================] - 0s 7ms/step - loss: 6563.8975 - accuracy: 0.5298\n",
            "Epoch 9/10\n",
            "7/7 [==============================] - 0s 8ms/step - loss: 6573.6104 - accuracy: 0.5131\n",
            "Epoch 10/10\n",
            "7/7 [==============================] - 0s 7ms/step - loss: 5070.7119 - accuracy: 0.5752\n"
          ]
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "<keras.callbacks.History at 0x7f64138b0790>"
            ]
          },
          "metadata": {},
          "execution_count": 157
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "test_ds = tfio.experimental.streaming.KafkaGroupIODataset(\n",
        "    topics=[\"cancer-test1\"],\n",
        "    group_id=\"testcg\",\n",
        "    servers=\"127.0.0.1:9092\",\n",
        "    stream_timeout=10000,\n",
        "    configuration=[\n",
        "        \"session.timeout.ms=7000\",\n",
        "        \"max.poll.interval.ms=8000\",\n",
        "        \"auto.offset.reset=earliest\"\n",
        "    ],\n",
        ")\n",
        "\n",
        "def decode_kafka_test_item(raw_message, raw_key):\n",
        "  message = tf.io.decode_csv(raw_message, [[0.0] for i in range(NUM_COLUMNS)])\n",
        "  key = tf.strings.to_number(raw_key)\n",
        "  return (message, key)\n",
        "\n",
        "test_ds = test_ds.map(decode_kafka_test_item)\n",
        "test_ds = test_ds.batch(BATCH_SIZE)"
      ],
      "metadata": {
        "id": "XNaP_4EYn1gc",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "2eb9d12d-2e82-4345-e2bf-ab91e551320c"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "WARNING:tensorflow:From /usr/local/lib/python3.8/dist-packages/tensorflow_io/python/experimental/kafka_group_io_dataset_ops.py:187: take_while (from tensorflow.python.data.experimental.ops.take_while_ops) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Use `tf.data.Dataset.take_while(...)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "res = model.evaluate(test_ds)\n",
        "print(\"test loss, test acc:\", res)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "RsgpYuYS6sCT",
        "outputId": "26d624c6-cded-4860-dc64-c9faaac26887"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "5/5 [==============================] - 11s 3s/step - loss: 2.1182 - accuracy: 0.3321\n",
            "test loss, test acc: [2.1182117462158203, 0.3321428596973419]\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "online_train_ds = tfio.experimental.streaming.KafkaBatchIODataset(\n",
        "    topics=[\"cancer-train1\"],\n",
        "    group_id=\"cgonline\",\n",
        "    servers=\"127.0.0.1:9092\",\n",
        "    stream_timeout=10000, # in milliseconds, to block indefinitely, set it to -1.\n",
        "    configuration=[\n",
        "        \"session.timeout.ms=7000\",\n",
        "        \"max.poll.interval.ms=8000\",\n",
        "        \"auto.offset.reset=earliest\"\n",
        "    ],\n",
        ")"
      ],
      "metadata": {
        "id": "8L_w8An86yMp",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "e4a88d32-12c3-4f7a-bdcc-a7cbdd0015e5"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "WARNING:tensorflow:From /usr/local/lib/python3.8/dist-packages/tensorflow_io/python/experimental/kafka_batch_io_dataset_ops.py:143: take_while (from tensorflow.python.data.experimental.ops.take_while_ops) is deprecated and will be removed in a future version.\n",
            "Instructions for updating:\n",
            "Use `tf.data.Dataset.take_while(...)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "def decode_kafka_online_item(raw_message, raw_key):\n",
        "  message = tf.io.decode_csv(raw_message, [[0.0] for i in range(NUM_COLUMNS)])\n",
        "  key = tf.strings.to_number(raw_key)\n",
        "  return (message, key)\n",
        "  \n",
        "for mini_ds in online_train_ds:\n",
        "  mini_ds = mini_ds.shuffle(buffer_size=32)\n",
        "  mini_ds = mini_ds.map(decode_kafka_online_item)\n",
        "  mini_ds = mini_ds.batch(32)\n",
        "  if len(mini_ds) > 0:\n",
        "    model.fit(mini_ds, epochs=3)"
      ],
      "metadata": {
        "id": "5Tb0IAyS7Xml",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "3f5e9782-27d8-4a1e-d4ee-9abc37d20e85"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Epoch 1/3\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/usr/local/lib/python3.8/dist-packages/tensorflow/python/util/dispatch.py:1096: UserWarning: \"`binary_crossentropy` received `from_logits=True`, but the `output` argument was produced by a sigmoid or softmax activation and thus does not represent logits. Was this intended?\"\n",
            "  return dispatch_target(*args, **kwargs)\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "14/14 [==============================] - 1s 5ms/step - loss: 26048.4512 - accuracy: 0.5274\n",
            "Epoch 2/3\n",
            "14/14 [==============================] - 0s 5ms/step - loss: 16865.8242 - accuracy: 0.5394\n",
            "Epoch 3/3\n",
            "14/14 [==============================] - 0s 6ms/step - loss: 13824.9336 - accuracy: 0.5871\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "G7Z68ggb7lVd"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}