python-jupyter-apache-kafka-ksql-tensorflow-keras.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Apache Kafka Integration + Preprocessing / Interactive Analysis with KSQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook uses the combination of Python, Apache Kafka, KSQL for Machine Learning infrastructures. \n",
    "\n",
    "It includes code examples using ksql-python and other widespread components from Python’s machine learning ecosystem, like Numpy, pandas, TensorFlow and Keras. \n",
    "\n",
    "The use case is fraud detection for credit card payments. We use a test data set from Kaggle as foundation to train an unsupervised autoencoder to detect anomalies and potential fraud in payments. Focus of this example is not just model training, but the whole Machine Learning infrastructure including data ingestion, data preprocessing, model training, model deployment and monitoring. All of this needs to be scalable, reliable and performant.\n",
    "\n",
    "If you want to learn more about the relation between the Apache Kafka open source ecosystem and Machine Learning, please check out these two blog posts:\n",
    "\n",
    "- [How to Build and Deploy Scalable Machine Learning in Production with Apache Kafka](https://www.confluent.io/blog/build-deploy-scalable-machine-learning-production-apache-kafka/)\n",
    "- [Using Apache Kafka to Drive Cutting-Edge Machine Learning](https://www.confluent.io/blog/using-apache-kafka-drive-cutting-edge-machine-learning)\n",
    "\n",
    "##### This notebook is not meant to be perfect using all coding and ML best practices, but just a simple guide how to build your own notebooks where you can combine Python APIs with Kafka and KSQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Start Backend Services (Zookeeper, Kafka, KSQL)\n",
    "\n",
    "The only server requirement is a local KSQL server running (with Kafka broker ZK node). If you don't have it running, just use Confluent CLI:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "This CLI is intended for development only, not for production\n",
      "https://docs.confluent.io/current/cli/index.html\n",
      "\n",
      "Using CONFLUENT_CURRENT: /var/folders/0s/0xdkb9n12yqdb3fs71926z3c0000gp/T/confluent.KQJxmlHC\n",
      "Starting zookeeper\n",
      "zookeeper is [\u001b[0;32mUP\u001b[0m]\n",
      "Starting kafka\n",
      "kafka is [\u001b[0;32mUP\u001b[0m]\n",
      "Starting schema-registry\n",
      "schema-registry is [\u001b[0;32mUP\u001b[0m]\n",
      "Starting ksql-server\n",
      "ksql-server is [\u001b[0;32mUP\u001b[0m]\n"
     ]
    }
   ],
   "source": [
    "# Shows correct startup but does not work 100% yet. Better run this command from outside Jupyter if you have any problems (e.g. from Terminal)!\n",
    "! confluent start ksql-server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Integration and Preprocessing with Python and KSQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First of all, create the Kafka Topic 'creditcardfraud_source' if it does not exist already:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.\n",
      "Created topic \"creditcardfraud_source\".\n"
     ]
    }
   ],
   "source": [
    "! kafka-topics --zookeeper localhost:2181 --create --topic creditcardfraud_source --partitions 3 --replication-factor 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then load KSQL library and initiate connection to KSQL server:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from ksql import KSQLAPI\n",
    "client = KSQLAPI('http://localhost:8088')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Consume source data from Kafka Topic \"creditcardfraud_source\":"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.create_stream(table_name='creditcardfraud_source',\n",
    "                     columns_type=['Id bigint', 'Timestamp varchar', 'User varchar', 'Time int', 'V1 double', 'V2 double', 'V3 double', 'V4 double', 'V5 double', 'V6 double', 'V7 double', 'V8 double', 'V9 double', 'V10 double', 'V11 double', 'V12 double', 'V13 double', 'V14 double', 'V15 double', 'V16 double', 'V17 double', 'V18 double', 'V19 double', 'V20 double', 'V21 double', 'V22 double', 'V23 double', 'V24 double', 'V25 double', 'V26 double', 'V27 double', 'V28 double', 'Amount double', 'Class string'],\n",
    "                     topic='creditcardfraud_source',\n",
    "                     value_format='DELIMITED')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Preprocessing: \n",
    "\n",
    "- Filter columns which are not needed \n",
    "- Filter messages where column 'class' is empty\n",
    "- Change data format to Avro for more convenient further processing\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.create_stream_as(table_name='creditcardfraud_preprocessed_avro',\n",
    "                     select_columns=['Time', 'V1', 'V2', 'V3', 'V4', 'V5', 'V6', 'V7', 'V8', 'V9', 'V10', 'V11', 'V12', 'V13', 'V14', 'V15', 'V16', 'V17', 'V18', 'V19', 'V20', 'V21', 'V22', 'V23', 'V24', 'V25', 'V26', 'V27', 'V28', 'Amount', 'Class'],\n",
    "                     src_table='creditcardfraud_source',\n",
    "                     conditions='Class IS NOT NULL',\n",
    "                     kafka_topic='creditcardfraud_preprocessed_avro',\n",
    "                     value_format='AVRO')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Take a look at the creates KSQL Streams:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'@type': 'streams',\n",
       "  'statementText': 'show streams;',\n",
       "  'streams': [{'type': 'STREAM',\n",
       "    'name': 'CREDITCARDFRAUD_SOURCE',\n",
       "    'topic': 'creditcardfraud_source',\n",
       "    'format': 'DELIMITED'},\n",
       "   {'type': 'STREAM',\n",
       "    'name': 'CREDITCARDFRAUD_PREPROCESSED_AVRO',\n",
       "    'topic': 'creditcardfraud_preprocessed_avro',\n",
       "    'format': 'AVRO'}]}]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.ksql('show streams')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Take a look at the metadata of the KSQL Stream:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'@type': 'sourceDescription',\n",
       "  'statementText': 'describe CREDITCARDFRAUD_PREPROCESSED_AVRO;',\n",
       "  'sourceDescription': {'name': 'CREDITCARDFRAUD_PREPROCESSED_AVRO',\n",
       "   'readQueries': [],\n",
       "   'writeQueries': [{'sinks': ['CREDITCARDFRAUD_PREPROCESSED_AVRO'],\n",
       "     'id': 'CSAS_CREDITCARDFRAUD_PREPROCESSED_AVRO_0',\n",
       "     'queryString': \"CREATE stream creditcardfraud_preprocessed_avro WITH (kafka_topic='creditcardfraud_preprocessed_avro', value_format='AVRO') AS SELECT Time, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12, V13, V14, V15, V16, V17, V18, V19, V20, V21, V22, V23, V24, V25, V26, V27, V28, Amount, Class FROM creditcardfraud_source where Class IS NOT NULL;\"}],\n",
       "   'fields': [{'name': 'ROWTIME',\n",
       "     'schema': {'type': 'BIGINT', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'ROWKEY',\n",
       "     'schema': {'type': 'STRING', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'TIME',\n",
       "     'schema': {'type': 'INTEGER', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V1',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V2',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V3',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V4',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V5',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V6',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V7',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V8',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V9',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V10',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V11',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V12',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V13',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V14',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V15',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V16',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V17',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V18',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V19',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V20',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V21',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V22',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V23',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V24',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V25',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V26',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V27',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'V28',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'AMOUNT',\n",
       "     'schema': {'type': 'DOUBLE', 'fields': None, 'memberSchema': None}},\n",
       "    {'name': 'CLASS',\n",
       "     'schema': {'type': 'STRING', 'fields': None, 'memberSchema': None}}],\n",
       "   'type': 'STREAM',\n",
       "   'key': '',\n",
       "   'timestamp': '',\n",
       "   'statistics': '',\n",
       "   'errorStats': '',\n",
       "   'extended': False,\n",
       "   'format': 'AVRO',\n",
       "   'topic': 'creditcardfraud_preprocessed_avro',\n",
       "   'partitions': 0,\n",
       "   'replication': 0}}]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.ksql('describe CREDITCARDFRAUD_PREPROCESSED_AVRO')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Interactive query statement:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "\n",
      "\n",
      "{\"row\":{\"columns\":[1547818491376,null,0,-1.3598071336738,-0.0727811733098497,2.53634673796914,1.37815522427443,-0.338320769942518,0\n",
      ".462387777762292,0.239598554061257,0.0986979012610507,0.363786969611213,0.0907941719789316,-0.551599533260813,-0.617800855762348,-0.991389847235408,-0.311169353699879,1.46817697209427,-0.470400525259478,0.207971241929242,0.0257905801985591,0.403992960255733,0.25141209823970\n",
      "5,-0.018306777944153,0.277837575558899,-0.110473910188767,0.0669280749146731,0.128539358273528,-0.189114843888824,0.133558376740387,-0.0210530534538215,149.62,\"0\"]},\"errorMessage\":null,\"finalMessage\":null}\n",
      "{\"row\":null,\"errorMessage\":null,\"finalMessage\":\"Limit Reached\"}\n",
      "\n"
     ]
    }
   ],
   "source": [
    "query = client.query('SELECT * FROM CREDITCARDFRAUD_PREPROCESSED_AVRO LIMIT 1')\n",
    "\n",
    "for item in query: \n",
    "    print(item)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Produce single test data manually (if you did not connect to a real data stream which produces data continuously), e.g. from terminal:\n",
    "\n",
    "                confluent produce creditcardfraud_source\n",
    "\n",
    "                1,\"2018-12- 18T12:00:00Z\",\"Hans\",0,-1.3598071336738,-0.0727811733098497,2.53634673796914,1.37815522427443,-0.338320769942518,0.462387777762292,0.239598554061257,0.0986979012610507,0.363786969611213,0.0907941719789316,-0.551599533260813,-0.617800855762348,-0.991389847235408,-0.311169353699879,1.46817697209427,-0.470400525259478,0.207971241929242,0.0257905801985591,0.403992960255733,0.251412098239705,-0.018306777944153,0.277837575558899,-0.110473910188767,0.0669280749146731,0.128539358273528,-0.189114843888824,0.133558376740387,-0.0210530534538215,149.62,\"0\"\n",
    "                \n",
    "*BE AWARE: The KSQL Python API does a REST call. This only waits a few seconds by default and then throws a timeout exception. You need to get data into the query before the timeout (e.g. by using above command).*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO How to embed ' ' in Python ???\n",
    "# See https://github.com/bryanyang0528/ksql-python/issues/54\n",
    "# client.ksql('SET \\'auto.offset.reset\\'=\\'earliest\\'');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Additional (optional) analysis and preprocessing examples\n",
    "\n",
    "Some more examples for possible data wrangling and preprocessing with KSQL:\n",
    "\n",
    "- Anonymization\n",
    "- Augmentation\n",
    "- Merge / Join data frames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "\n",
      "{\"row\":{\"columns\":[1,\"Xxns\"]},\"errorMessage\":null,\"finalMessage\":null}\n",
      "{\"row\":null,\"errorMessage\":null,\"finalMessage\":\"Limit Reached\"}\n",
      "\n"
     ]
    }
   ],
   "source": [
    "query = client.query('SELECT Id, MASK_LEFT(User, 2) FROM creditcardfraud_source LIMIT 1')\n",
    "\n",
    "for item in query: \n",
    "    print(item)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "{\"row\":{\"columns\":[1,\"-1\"]},\"errorMessage\":null,\"finalMessage\":null}\n",
      "{\"row\":null,\"errorMessage\":null,\"finalMessage\":\"Limit Reached\"}\n",
      "\n"
     ]
    }
   ],
   "source": [
    "query = client.query('SELECT Id, IFNULL(Class, \\'-1\\') FROM creditcardfraud_source LIMIT 1')\n",
    "\n",
    "for item in query: \n",
    "    print(item)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Stream-Table-Join\n",
    "\n",
    "For the STREAM-TABLE-JOIN, you first need to create a Kafka Topic 'Users' (for the corresponding KSQL TABLE 'Users):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Created topic \"users\".\r\n"
     ]
    }
   ],
   "source": [
    "! kafka-topics --zookeeper localhost:2181 --create --topic users --partitions 3 --replication-factor 1 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then create the KSQL Table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.create_table(table_name='users',\n",
    "                     columns_type=['userid varchar', 'gender varchar', 'regionid varchar'],\n",
    "                     topic='users',\n",
    "                     key='userid',\n",
    "                     value_format='AVRO')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'@type': 'currentStatus',\n",
       "  'statementText': \"CREATE STREAM creditcardfraud_per_user WITH (VALUE_FORMAT='AVRO', KAFKA_TOPIC='creditcardfraud_per_user') AS SELECT Time, Amount, Class FROM creditcardfraud_source c INNER JOIN USERS u on c.user = u.userid WHERE u.USERID = 1;\",\n",
       "  'commandId': 'stream/CREDITCARDFRAUD_PER_USER/create',\n",
       "  'commandStatus': {'status': 'SUCCESS',\n",
       "   'message': 'Stream created and running'}}]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.ksql(\"CREATE STREAM creditcardfraud_per_user WITH (VALUE_FORMAT='AVRO', KAFKA_TOPIC='creditcardfraud_per_user') AS SELECT Time, Amount, Class FROM creditcardfraud_source c INNER JOIN USERS u on c.user = u.userid WHERE u.USERID = 1\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Mapping from KSQL to NumPy / pandas for Machine Learning tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "import json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The query below command returns a Python generator. It can be printed e.g. by reading its values via next(query) or a for loop.\n",
    "\n",
    "Due to a current [bug in ksql-python library](https://github.com/bryanyang0528/ksql-python/issues/57), we need to to an additional line of Python code to strip out unnecessary info and change to 2D array "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<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>Time</th>\n",
       "      <th>V1</th>\n",
       "      <th>V2</th>\n",
       "      <th>V3</th>\n",
       "      <th>V4</th>\n",
       "      <th>V5</th>\n",
       "      <th>V6</th>\n",
       "      <th>V7</th>\n",
       "      <th>V8</th>\n",
       "      <th>V9</th>\n",
       "      <th>...</th>\n",
       "      <th>V21</th>\n",
       "      <th>V22</th>\n",
       "      <th>V23</th>\n",
       "      <th>V24</th>\n",
       "      <th>V25</th>\n",
       "      <th>V26</th>\n",
       "      <th>V27</th>\n",
       "      <th>V28</th>\n",
       "      <th>Amount</th>\n",
       "      <th>Class</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2</td>\n",
       "      <td>-0.425966</td>\n",
       "      <td>0.960523</td>\n",
       "      <td>1.141109</td>\n",
       "      <td>-0.168252</td>\n",
       "      <td>0.420987</td>\n",
       "      <td>-0.029728</td>\n",
       "      <td>0.476201</td>\n",
       "      <td>0.260314</td>\n",
       "      <td>-0.568671</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.208254</td>\n",
       "      <td>-0.559825</td>\n",
       "      <td>-0.026398</td>\n",
       "      <td>-0.371427</td>\n",
       "      <td>-0.232794</td>\n",
       "      <td>0.105915</td>\n",
       "      <td>0.253844</td>\n",
       "      <td>0.081080</td>\n",
       "      <td>3.67</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2</td>\n",
       "      <td>-1.158233</td>\n",
       "      <td>0.877737</td>\n",
       "      <td>1.548718</td>\n",
       "      <td>0.403034</td>\n",
       "      <td>-0.407193</td>\n",
       "      <td>0.095921</td>\n",
       "      <td>0.592941</td>\n",
       "      <td>-0.270533</td>\n",
       "      <td>0.817739</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.009431</td>\n",
       "      <td>0.798278</td>\n",
       "      <td>-0.137458</td>\n",
       "      <td>0.141267</td>\n",
       "      <td>-0.206010</td>\n",
       "      <td>0.502292</td>\n",
       "      <td>0.219422</td>\n",
       "      <td>0.215153</td>\n",
       "      <td>69.99</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>7</td>\n",
       "      <td>-0.894286</td>\n",
       "      <td>0.286157</td>\n",
       "      <td>-0.113192</td>\n",
       "      <td>-0.271526</td>\n",
       "      <td>2.669599</td>\n",
       "      <td>3.721818</td>\n",
       "      <td>0.370145</td>\n",
       "      <td>0.851084</td>\n",
       "      <td>-0.392048</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.073425</td>\n",
       "      <td>-0.268092</td>\n",
       "      <td>-0.204233</td>\n",
       "      <td>1.011592</td>\n",
       "      <td>0.373205</td>\n",
       "      <td>-0.384157</td>\n",
       "      <td>0.011747</td>\n",
       "      <td>0.142404</td>\n",
       "      <td>93.20</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>7</td>\n",
       "      <td>-0.644269</td>\n",
       "      <td>1.417964</td>\n",
       "      <td>1.074380</td>\n",
       "      <td>-0.492199</td>\n",
       "      <td>0.948934</td>\n",
       "      <td>0.428118</td>\n",
       "      <td>1.120631</td>\n",
       "      <td>-3.807864</td>\n",
       "      <td>0.615375</td>\n",
       "      <td>...</td>\n",
       "      <td>1.943465</td>\n",
       "      <td>-1.015455</td>\n",
       "      <td>0.057504</td>\n",
       "      <td>-0.649709</td>\n",
       "      <td>-0.415267</td>\n",
       "      <td>-0.051634</td>\n",
       "      <td>-1.206921</td>\n",
       "      <td>-1.085339</td>\n",
       "      <td>40.80</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>0</td>\n",
       "      <td>1.191857</td>\n",
       "      <td>0.266151</td>\n",
       "      <td>0.166480</td>\n",
       "      <td>0.448154</td>\n",
       "      <td>0.060018</td>\n",
       "      <td>-0.082361</td>\n",
       "      <td>-0.078803</td>\n",
       "      <td>0.085102</td>\n",
       "      <td>-0.255425</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.225775</td>\n",
       "      <td>-0.638672</td>\n",
       "      <td>0.101288</td>\n",
       "      <td>-0.339846</td>\n",
       "      <td>0.167170</td>\n",
       "      <td>0.125895</td>\n",
       "      <td>-0.008983</td>\n",
       "      <td>0.014724</td>\n",
       "      <td>2.69</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>1</td>\n",
       "      <td>-0.966272</td>\n",
       "      <td>-0.185226</td>\n",
       "      <td>1.792993</td>\n",
       "      <td>-0.863291</td>\n",
       "      <td>-0.010309</td>\n",
       "      <td>1.247203</td>\n",
       "      <td>0.237609</td>\n",
       "      <td>0.377436</td>\n",
       "      <td>-1.387024</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.108300</td>\n",
       "      <td>0.005274</td>\n",
       "      <td>-0.190321</td>\n",
       "      <td>-1.175575</td>\n",
       "      <td>0.647376</td>\n",
       "      <td>-0.221929</td>\n",
       "      <td>0.062723</td>\n",
       "      <td>0.061458</td>\n",
       "      <td>123.50</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>0</td>\n",
       "      <td>-1.359807</td>\n",
       "      <td>-0.072781</td>\n",
       "      <td>2.536347</td>\n",
       "      <td>1.378155</td>\n",
       "      <td>-0.338321</td>\n",
       "      <td>0.462388</td>\n",
       "      <td>0.239599</td>\n",
       "      <td>0.098698</td>\n",
       "      <td>0.363787</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.018307</td>\n",
       "      <td>0.277838</td>\n",
       "      <td>-0.110474</td>\n",
       "      <td>0.066928</td>\n",
       "      <td>0.128539</td>\n",
       "      <td>-0.189115</td>\n",
       "      <td>0.133558</td>\n",
       "      <td>-0.021053</td>\n",
       "      <td>149.62</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>1</td>\n",
       "      <td>-1.358354</td>\n",
       "      <td>-1.340163</td>\n",
       "      <td>1.773209</td>\n",
       "      <td>0.379780</td>\n",
       "      <td>-0.503198</td>\n",
       "      <td>1.800499</td>\n",
       "      <td>0.791461</td>\n",
       "      <td>0.247676</td>\n",
       "      <td>-1.514654</td>\n",
       "      <td>...</td>\n",
       "      <td>0.247998</td>\n",
       "      <td>0.771679</td>\n",
       "      <td>0.909412</td>\n",
       "      <td>-0.689281</td>\n",
       "      <td>-0.327642</td>\n",
       "      <td>-0.139097</td>\n",
       "      <td>-0.055353</td>\n",
       "      <td>-0.059752</td>\n",
       "      <td>378.66</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>8 rows × 31 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   Time        V1        V2        V3        V4        V5        V6        V7  \\\n",
       "0     2 -0.425966  0.960523  1.141109 -0.168252  0.420987 -0.029728  0.476201   \n",
       "1     2 -1.158233  0.877737  1.548718  0.403034 -0.407193  0.095921  0.592941   \n",
       "2     7 -0.894286  0.286157 -0.113192 -0.271526  2.669599  3.721818  0.370145   \n",
       "3     7 -0.644269  1.417964  1.074380 -0.492199  0.948934  0.428118  1.120631   \n",
       "4     0  1.191857  0.266151  0.166480  0.448154  0.060018 -0.082361 -0.078803   \n",
       "5     1 -0.966272 -0.185226  1.792993 -0.863291 -0.010309  1.247203  0.237609   \n",
       "6     0 -1.359807 -0.072781  2.536347  1.378155 -0.338321  0.462388  0.239599   \n",
       "7     1 -1.358354 -1.340163  1.773209  0.379780 -0.503198  1.800499  0.791461   \n",
       "\n",
       "         V8        V9  ...         V21       V22       V23       V24  \\\n",
       "0  0.260314 -0.568671  ...   -0.208254 -0.559825 -0.026398 -0.371427   \n",
       "1 -0.270533  0.817739  ...   -0.009431  0.798278 -0.137458  0.141267   \n",
       "2  0.851084 -0.392048  ...   -0.073425 -0.268092 -0.204233  1.011592   \n",
       "3 -3.807864  0.615375  ...    1.943465 -1.015455  0.057504 -0.649709   \n",
       "4  0.085102 -0.255425  ...   -0.225775 -0.638672  0.101288 -0.339846   \n",
       "5  0.377436 -1.387024  ...   -0.108300  0.005274 -0.190321 -1.175575   \n",
       "6  0.098698  0.363787  ...   -0.018307  0.277838 -0.110474  0.066928   \n",
       "7  0.247676 -1.514654  ...    0.247998  0.771679  0.909412 -0.689281   \n",
       "\n",
       "        V25       V26       V27       V28  Amount  Class  \n",
       "0 -0.232794  0.105915  0.253844  0.081080    3.67      0  \n",
       "1 -0.206010  0.502292  0.219422  0.215153   69.99      1  \n",
       "2  0.373205 -0.384157  0.011747  0.142404   93.20      1  \n",
       "3 -0.415267 -0.051634 -1.206921 -1.085339   40.80      0  \n",
       "4  0.167170  0.125895 -0.008983  0.014724    2.69      0  \n",
       "5  0.647376 -0.221929  0.062723  0.061458  123.50      0  \n",
       "6  0.128539 -0.189115  0.133558 -0.021053  149.62      0  \n",
       "7 -0.327642 -0.139097 -0.055353 -0.059752  378.66      0  \n",
       "\n",
       "[8 rows x 31 columns]"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = client.query('SELECT * FROM CREDITCARDFRAUD_PREPROCESSED_AVRO LIMIT 8') # Returns a Python generator object\n",
    "\n",
    "#items = [item for item in query][:-1]        # -1 to remove last record that is a dummy msg for \"Limit Reached\"          \n",
    "#one_record = json.loads(''.join(items))      # Join two records as one as ksql-python is splitting it into two?          \n",
    "#data = [one_record['row']['columns'][2:-1]]  # Strip out unnecessary info and change to 2D array                     \n",
    "#df = pd.DataFrame(data=data)   \n",
    "\n",
    "records = [json.loads(r) for r in ''.join(query).strip().replace('\\n\\n\\n\\n', '').split('\\n')]\n",
    "data = [r['row']['columns'][2:] for r in records[:-1]]\n",
    "#data = r['row']['columns'][2] for r in records\n",
    "df = pd.DataFrame(data=data, columns=['Time',  'V1' , 'V2' , 'V3' , 'V4' , 'V5' , 'V6' , 'V7' , 'V8' , 'V9' , 'V10' , 'V11' , 'V12' , 'V13' , 'V14' , 'V15' , 'V16' , 'V17' , 'V18' , 'V19' , 'V20' , 'V21' , 'V22' , 'V23' , 'V24' , 'V25' , 'V26' , 'V27' , 'V28' , 'Amount' , 'Class'])\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generate some test data \n",
    "\n",
    "As discussed in the step-by-step guide, you have various options. Here we - ironically - read messages from a CSV file. This is for simple demo purposes so that you don't have to set up a real continuous Kafka stream. \n",
    "\n",
    "In real world or more advanced examples, you should connect to a real Kafka data stream (for instance using the Kafka data generator or Kafka Connect).\n",
    "\n",
    "Here we just consume a few messages for demo purposes so that they get mapped into a pandas dataframe:\n",
    "\n",
    "                cat /Users/kai.waehner/git-projects/python-jupyter-apache-kafka-ksql-tensorflow-keras/data/creditcard_extended.csv | kafka-console-producer --broker-list localhost:9092 --topic creditcardfraud_source\n",
    "                \n",
    "You need to do this from command line because Jupyter cannot execute this in parallel to above KSQL query."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Preprocessing with Pandas + Model Training with TensorFlow / Keras"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### BE AWARE: You need enough messages in the pandas data frame to train the model in the below cells (if you just play around with ksql-python and just add a few Kafka events, it is not a sufficient number of rows to continue. You can simply change to df = pd.read_csv(\"data/creditcard.csv\") as shown below in this case to get a bigger data set...\n",
    "\n",
    "\n",
    "This part only includes the steps required for model training of the Autoencoder with Keras and TensorFlow. \n",
    "\n",
    "If you want to get a better understanding of the model, take a look at the other notebook [Python Tensorflow Keras Fraud Detection Autoencoder.ipynb](http://localhost:8888/notebooks/Python%20Tensorflow%20Keras%20Fraud%20Detection%20Autoencoder.ipynb) which includes many more details, plots and explanations.\n",
    "\n",
    "[Kudos to David Ellison](https://www.datascience.com/blog/fraud-detection-with-tensorflow).\n",
    "\n",
    "[The credit card fraud data set is available at Kaggle](https://www.kaggle.com/mlg-ulb/creditcardfraud/data)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Using TensorFlow backend.\n"
     ]
    }
   ],
   "source": [
    "# import packages\n",
    "# matplotlib inline\n",
    "#import pandas as pd\n",
    "#import numpy as np\n",
    "from scipy import stats\n",
    "import tensorflow as tf\n",
    "import matplotlib.pyplot as plt\n",
    "import seaborn as sns\n",
    "import pickle\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.metrics import confusion_matrix, precision_recall_curve\n",
    "from sklearn.metrics import recall_score, classification_report, auc, roc_curve\n",
    "from sklearn.metrics import precision_recall_fscore_support, f1_score\n",
    "from sklearn.preprocessing import StandardScaler\n",
    "from pylab import rcParams\n",
    "from keras.models import Model, load_model\n",
    "from keras.layers import Input, Dense\n",
    "from keras.callbacks import ModelCheckpoint, TensorBoard\n",
    "from keras import regularizers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<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>Time</th>\n",
       "      <th>V1</th>\n",
       "      <th>V2</th>\n",
       "      <th>V3</th>\n",
       "      <th>V4</th>\n",
       "      <th>V5</th>\n",
       "      <th>V6</th>\n",
       "      <th>V7</th>\n",
       "      <th>V8</th>\n",
       "      <th>V9</th>\n",
       "      <th>...</th>\n",
       "      <th>V21</th>\n",
       "      <th>V22</th>\n",
       "      <th>V23</th>\n",
       "      <th>V24</th>\n",
       "      <th>V25</th>\n",
       "      <th>V26</th>\n",
       "      <th>V27</th>\n",
       "      <th>V28</th>\n",
       "      <th>Amount</th>\n",
       "      <th>Class</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0.0</td>\n",
       "      <td>-1.359807</td>\n",
       "      <td>-0.072781</td>\n",
       "      <td>2.536347</td>\n",
       "      <td>1.378155</td>\n",
       "      <td>-0.338321</td>\n",
       "      <td>0.462388</td>\n",
       "      <td>0.239599</td>\n",
       "      <td>0.098698</td>\n",
       "      <td>0.363787</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.018307</td>\n",
       "      <td>0.277838</td>\n",
       "      <td>-0.110474</td>\n",
       "      <td>0.066928</td>\n",
       "      <td>0.128539</td>\n",
       "      <td>-0.189115</td>\n",
       "      <td>0.133558</td>\n",
       "      <td>-0.021053</td>\n",
       "      <td>149.62</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0.0</td>\n",
       "      <td>1.191857</td>\n",
       "      <td>0.266151</td>\n",
       "      <td>0.166480</td>\n",
       "      <td>0.448154</td>\n",
       "      <td>0.060018</td>\n",
       "      <td>-0.082361</td>\n",
       "      <td>-0.078803</td>\n",
       "      <td>0.085102</td>\n",
       "      <td>-0.255425</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.225775</td>\n",
       "      <td>-0.638672</td>\n",
       "      <td>0.101288</td>\n",
       "      <td>-0.339846</td>\n",
       "      <td>0.167170</td>\n",
       "      <td>0.125895</td>\n",
       "      <td>-0.008983</td>\n",
       "      <td>0.014724</td>\n",
       "      <td>2.69</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1.0</td>\n",
       "      <td>-1.358354</td>\n",
       "      <td>-1.340163</td>\n",
       "      <td>1.773209</td>\n",
       "      <td>0.379780</td>\n",
       "      <td>-0.503198</td>\n",
       "      <td>1.800499</td>\n",
       "      <td>0.791461</td>\n",
       "      <td>0.247676</td>\n",
       "      <td>-1.514654</td>\n",
       "      <td>...</td>\n",
       "      <td>0.247998</td>\n",
       "      <td>0.771679</td>\n",
       "      <td>0.909412</td>\n",
       "      <td>-0.689281</td>\n",
       "      <td>-0.327642</td>\n",
       "      <td>-0.139097</td>\n",
       "      <td>-0.055353</td>\n",
       "      <td>-0.059752</td>\n",
       "      <td>378.66</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1.0</td>\n",
       "      <td>-0.966272</td>\n",
       "      <td>-0.185226</td>\n",
       "      <td>1.792993</td>\n",
       "      <td>-0.863291</td>\n",
       "      <td>-0.010309</td>\n",
       "      <td>1.247203</td>\n",
       "      <td>0.237609</td>\n",
       "      <td>0.377436</td>\n",
       "      <td>-1.387024</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.108300</td>\n",
       "      <td>0.005274</td>\n",
       "      <td>-0.190321</td>\n",
       "      <td>-1.175575</td>\n",
       "      <td>0.647376</td>\n",
       "      <td>-0.221929</td>\n",
       "      <td>0.062723</td>\n",
       "      <td>0.061458</td>\n",
       "      <td>123.50</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2.0</td>\n",
       "      <td>-1.158233</td>\n",
       "      <td>0.877737</td>\n",
       "      <td>1.548718</td>\n",
       "      <td>0.403034</td>\n",
       "      <td>-0.407193</td>\n",
       "      <td>0.095921</td>\n",
       "      <td>0.592941</td>\n",
       "      <td>-0.270533</td>\n",
       "      <td>0.817739</td>\n",
       "      <td>...</td>\n",
       "      <td>-0.009431</td>\n",
       "      <td>0.798278</td>\n",
       "      <td>-0.137458</td>\n",
       "      <td>0.141267</td>\n",
       "      <td>-0.206010</td>\n",
       "      <td>0.502292</td>\n",
       "      <td>0.219422</td>\n",
       "      <td>0.215153</td>\n",
       "      <td>69.99</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 31 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "   Time        V1        V2        V3        V4        V5        V6        V7  \\\n",
       "0   0.0 -1.359807 -0.072781  2.536347  1.378155 -0.338321  0.462388  0.239599   \n",
       "1   0.0  1.191857  0.266151  0.166480  0.448154  0.060018 -0.082361 -0.078803   \n",
       "2   1.0 -1.358354 -1.340163  1.773209  0.379780 -0.503198  1.800499  0.791461   \n",
       "3   1.0 -0.966272 -0.185226  1.792993 -0.863291 -0.010309  1.247203  0.237609   \n",
       "4   2.0 -1.158233  0.877737  1.548718  0.403034 -0.407193  0.095921  0.592941   \n",
       "\n",
       "         V8        V9  ...         V21       V22       V23       V24  \\\n",
       "0  0.098698  0.363787  ...   -0.018307  0.277838 -0.110474  0.066928   \n",
       "1  0.085102 -0.255425  ...   -0.225775 -0.638672  0.101288 -0.339846   \n",
       "2  0.247676 -1.514654  ...    0.247998  0.771679  0.909412 -0.689281   \n",
       "3  0.377436 -1.387024  ...   -0.108300  0.005274 -0.190321 -1.175575   \n",
       "4 -0.270533  0.817739  ...   -0.009431  0.798278 -0.137458  0.141267   \n",
       "\n",
       "        V25       V26       V27       V28  Amount  Class  \n",
       "0  0.128539 -0.189115  0.133558 -0.021053  149.62      0  \n",
       "1  0.167170  0.125895 -0.008983  0.014724    2.69      0  \n",
       "2 -0.327642 -0.139097 -0.055353 -0.059752  378.66      0  \n",
       "3  0.647376 -0.221929  0.062723  0.061458  123.50      0  \n",
       "4 -0.206010  0.502292  0.219422  0.215153   69.99      0  \n",
       "\n",
       "[5 rows x 31 columns]"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Use the dataframe from above (imported and preprocessed with KSQL)\n",
    "\n",
    "# As alternative directly import from a CSV file (\"the normal approach without Kafka and streaming data\")\n",
    "\n",
    "# \"data/creditcard_small.csv\" is a very small data set (just for quick demo purpose to get a model binary)\n",
    "# => replace with \"data/creditcard.csv\" to use a real data set to train a model with good accuracy\n",
    "#df = pd.read_csv(\"data/creditcard.csv\") \n",
    "\n",
    "\n",
    "df.head(n=5) #just to check you imported the dataset properly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "#set random seed and percentage of test data\n",
    "RANDOM_SEED = 314 #used to help randomly select the data points\n",
    "TEST_PCT = 0.2 # 20% of the data\n",
    "\n",
    "#set up graphic style in this case I am using the color scheme from xkcd.com\n",
    "rcParams['figure.figsize'] = 14, 8.7 # Golden Mean\n",
    "LABELS = [\"Normal\",\"Fraud\"]\n",
    "#col_list = [\"cerulean\",\"scarlet\"]# https://xkcd.com/color/rgb/\n",
    "#sns.set(style='white', font_scale=1.75, palette=sns.xkcd_palette(col_list))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "normal_df = [df.Class == 0] #save normal_df observations into a separate df\n",
    "fraud_df = [df.Class == 1] #do the same for frauds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "#data = df.drop(['Time'], axis=1) #if you think the var is unimportant\n",
    "df_norm = df\n",
    "df_norm['Time'] = StandardScaler().fit_transform(df_norm['Time'].values.reshape(-1, 1))\n",
    "df_norm['Amount'] = StandardScaler().fit_transform(df_norm['Amount'].values.reshape(-1, 1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_x, test_x = train_test_split(df_norm, test_size=TEST_PCT, random_state=RANDOM_SEED)\n",
    "train_x = train_x[train_x.Class == 0] #where normal transactions\n",
    "train_x = train_x.drop(['Class'], axis=1) #drop the class column\n",
    "\n",
    "test_y = test_x['Class'] #save the class column for the test set\n",
    "test_x = test_x.drop(['Class'], axis=1) #drop the class column\n",
    "\n",
    "train_x = train_x.values #transform to ndarray\n",
    "test_x = test_x.values"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### My Jupyter Notebook crashed sometimes in the next step 'model training' (probably memory issues):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reduce number of epochs and batch_size if your Jupyter crashes (due to memory issues)\n",
    "# nb_epoch = 100\n",
    "# batch_size = 128\n",
    "nb_epoch = 5\n",
    "batch_size = 32\n",
    "\n",
    "input_dim = train_x.shape[1] #num of columns, 30\n",
    "encoding_dim = 14\n",
    "hidden_dim = int(encoding_dim / 2) #i.e. 7\n",
    "learning_rate = 1e-7\n",
    "\n",
    "input_layer = Input(shape=(input_dim, ))\n",
    "encoder = Dense(encoding_dim, activation=\"tanh\", activity_regularizer=regularizers.l1(learning_rate))(input_layer)\n",
    "encoder = Dense(hidden_dim, activation=\"relu\")(encoder)\n",
    "decoder = Dense(hidden_dim, activation='tanh')(encoder)\n",
    "decoder = Dense(input_dim, activation='relu')(decoder)\n",
    "autoencoder = Model(inputs=input_layer, outputs=decoder)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Train on 227468 samples, validate on 56962 samples\n",
      "Epoch 1/5\n"
     ]
    }
   ],
   "source": [
    "autoencoder.compile(metrics=['accuracy'],\n",
    "                    loss='mean_squared_error',\n",
    "                    optimizer='adam')\n",
    "\n",
    "cp = ModelCheckpoint(filepath=\"models/autoencoder_fraud.h5\",\n",
    "                               save_best_only=True,\n",
    "                               verbose=0)\n",
    "\n",
    "tb = TensorBoard(log_dir='./logs',\n",
    "                histogram_freq=0,\n",
    "                write_graph=True,\n",
    "                write_images=True)\n",
    "\n",
    "history = autoencoder.fit(train_x, train_x,\n",
    "                    epochs=nb_epoch,\n",
    "                    batch_size=batch_size,\n",
    "                    shuffle=True,\n",
    "                    validation_data=(test_x, test_x),\n",
    "                    verbose=1,\n",
    "                    callbacks=[cp, tb]).history"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "autoencoder = load_model('models/autoencoder_fraud.h5')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_x_predictions = autoencoder.predict(test_x)\n",
    "mse = np.mean(np.power(test_x - test_x_predictions, 2), axis=1)\n",
    "error_df = pd.DataFrame({'Reconstruction_error': mse,\n",
    "                        'True_class': test_y})\n",
    "error_df.describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The binary 'models/autoencoder_fraud.h5' is the trained model which can then be deployed anywhere to do prediction on new incoming events in real time. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model Deployment\n",
    "\n",
    "This demo focuses on the combination of Python and KSQL for data preprocessing and model training. If you want to understand the relation between Apache Kafka, KSQL and Python-related Machine Learning tools like TensorFlow for model deployment and monitoring, please check out my other Github projects:\n",
    "\n",
    "Some examples of model deployment in Kafka environments:\n",
    "\n",
    "- [Analytic models (TensorFlow, Keras, H2O and Deeplearning4j) embedded in Kafka Streams microservices](https://github.com/kaiwaehner/kafka-streams-machine-learning-examples)\n",
    "- [Anomaly detection of IoT sensor data with a model embedded into a KSQL UDF](https://github.com/kaiwaehner/ksql-udf-deep-learning-mqtt-iot)\n",
    "- [RPC communication between Kafka Streams application and model server (TensorFlow Serving)](https://github.com/kaiwaehner/tensorflow-serving-java-grpc-kafka-streams)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Appendix: Pandas analysis with above Fraud Detection Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.read_csv(\"data/creditcard.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df['Amount']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df[0:3]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.iloc[1,1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Takes a minute or two (big CSV file)...\n",
    "#df.plot()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}