Processing Continuously Coming Data  Using File Connector and in Mule 4-Part 3.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import Image\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import random\n",
    "import string\n",
    "import time\n",
    "from datetime import datetime\n",
    "import pymysql\n",
    "from IPython.display import Image\n",
    "from sqlalchemy import create_engine\n",
    "def generate_random(size):\n",
    "       return ''.join(random.choice(string.ascii_letters) for x in range(size))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](insert_2_mysql_continous_data_parallel.PNG)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](scatter-gather.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[YouTube Video Tutorial](https://youtu.be/TOEbsPBOOyI)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a database a table in MySQL\n",
    "\n",
    "CREATE DATABASE mulesoft;\n",
    "\n",
    "USE mulesoft;\n",
    "\n",
    "CREATE TABLE Persons (\n",
    "    PersonID int,\n",
    "    \n",
    "    LastName varchar(255),\n",
    "    \n",
    "    FirstName varchar(255),\n",
    "    \n",
    "    Address varchar(255),\n",
    "    \n",
    "    City varchar(255)\n",
    ");"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "### SQL query string\n",
    "\n",
    "INSERT INTO Persons (PersonID,LastName, FirstName, Address, City)\n",
    "\n",
    "VALUES (:PersonID, :LastName, :FirstName, :Address, :City)\n",
    "\n",
    "\n",
    "### Input parameters\n",
    "\n",
    "{\"PersonID\":payload.PersonID,\n",
    "\n",
    " \"LastName\":payload.LastName,\n",
    " \n",
    " \"FirstName\":payload.FirstName,\n",
    " \n",
    " \"Address\": payload.Address,\n",
    " \n",
    " \"City\":payload.City}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## PostgreSQL Config\n",
    "#### URL: jdbc:postgresql://localhost:5432/db_name\n",
    "#### Driver class name: org.postgresql.Driver"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_rows = 10\n",
    "num_characters = 4\n",
    "PersonID =  np.random.randint(200000, size=10)\n",
    "LastName =  [generate_random(num_characters) for x in range(num_rows)]\n",
    "FirstName = [generate_random(num_characters) for x in range(num_rows)]\n",
    "Address =   [generate_random(num_characters) for x in range(num_rows)]\n",
    "City =     [generate_random(num_characters) for x in range(num_rows)]\n",
    "df = pd.DataFrame({'PersonID':PersonID, 'LastName':LastName, 'FirstName':FirstName, 'Address':Address, 'City':City})\n",
    "df.to_json('tmp.json', orient ='records')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Continuously genetate and save csv file and png plot evey second"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_rows = 10\n",
    "num_characters = 4\n",
    "while True:\n",
    "    PersonID =  np.random.randint(200000, size=10)\n",
    "    LastName =  [generate_random(num_characters) for x in range(num_rows)]\n",
    "    FirstName = [generate_random(num_characters) for x in range(num_rows)]\n",
    "    Address =   [generate_random(num_characters) for x in range(num_rows)]\n",
    "    City =     [generate_random(num_characters) for x in range(num_rows)]\n",
    "    df = pd.DataFrame({'PersonID':PersonID, 'LastName':LastName, 'FirstName':FirstName, 'Address':Address, 'City':City})\n",
    "    file_name = 'data/' + str(datetime.now()).replace(' ', '_').replace(\":\", '_')[:19]\n",
    "    df.to_csv(file_name + '.csv', index = False)\n",
    "    plt.plot(np.random.randint(low = 0, high = 100, size = 20))\n",
    "    plt.savefig(file_name + 'png')\n",
    "    time.sleep(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Check data is being inserted into the databases"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "MySQL number of rows:  189\n",
      "PostgreSQL number of rows:  212\n",
      "MySQL number of rows:  217\n",
      "PostgreSQL number of rows:  242\n",
      "MySQL number of rows:  248\n",
      "PostgreSQL number of rows:  261\n",
      "MySQL number of rows:  265\n",
      "PostgreSQL number of rows:  270\n",
      "MySQL number of rows:  273\n",
      "PostgreSQL number of rows:  277\n",
      "MySQL number of rows:  284\n",
      "PostgreSQL number of rows:  287\n",
      "MySQL number of rows:  293\n",
      "PostgreSQL number of rows:  295\n",
      "MySQL number of rows:  299\n",
      "PostgreSQL number of rows:  303\n",
      "MySQL number of rows:  306\n",
      "PostgreSQL number of rows:  309\n",
      "MySQL number of rows:  312\n",
      "PostgreSQL number of rows:  313\n",
      "MySQL number of rows:  316\n",
      "PostgreSQL number of rows:  315\n",
      "MySQL number of rows:  320\n",
      "PostgreSQL number of rows:  322\n",
      "MySQL number of rows:  327\n",
      "PostgreSQL number of rows:  329\n",
      "MySQL number of rows:  331\n",
      "PostgreSQL number of rows:  334\n",
      "MySQL number of rows:  339\n",
      "PostgreSQL number of rows:  341\n",
      "MySQL number of rows:  345\n",
      "PostgreSQL number of rows:  348\n",
      "MySQL number of rows:  350\n",
      "PostgreSQL number of rows:  353\n",
      "MySQL number of rows:  359\n",
      "PostgreSQL number of rows:  360\n",
      "MySQL number of rows:  365\n",
      "PostgreSQL number of rows:  368\n",
      "MySQL number of rows:  372\n",
      "PostgreSQL number of rows:  374\n",
      "MySQL number of rows:  376\n",
      "PostgreSQL number of rows:  378\n",
      "MySQL number of rows:  382\n",
      "PostgreSQL number of rows:  385\n",
      "MySQL number of rows:  391\n",
      "PostgreSQL number of rows:  395\n",
      "MySQL number of rows:  399\n",
      "PostgreSQL number of rows:  404\n",
      "MySQL number of rows:  407\n",
      "PostgreSQL number of rows:  410\n",
      "MySQL number of rows:  414\n",
      "PostgreSQL number of rows:  417\n",
      "MySQL number of rows:  420\n",
      "PostgreSQL number of rows:  422\n",
      "MySQL number of rows:  428\n",
      "PostgreSQL number of rows:  429\n",
      "MySQL number of rows:  434\n",
      "PostgreSQL number of rows:  437\n",
      "MySQL number of rows:  442\n",
      "PostgreSQL number of rows:  442\n",
      "MySQL number of rows:  447\n",
      "PostgreSQL number of rows:  450\n",
      "MySQL number of rows:  454\n",
      "PostgreSQL number of rows:  457\n",
      "MySQL number of rows:  461\n",
      "PostgreSQL number of rows:  464\n",
      "MySQL number of rows:  468\n",
      "PostgreSQL number of rows:  471\n",
      "MySQL number of rows:  475\n",
      "PostgreSQL number of rows:  478\n",
      "MySQL number of rows:  483\n",
      "PostgreSQL number of rows:  486\n",
      "MySQL number of rows:  489\n",
      "PostgreSQL number of rows:  492\n",
      "MySQL number of rows:  494\n",
      "PostgreSQL number of rows:  496\n",
      "MySQL number of rows:  496\n",
      "PostgreSQL number of rows:  496\n",
      "MySQL number of rows:  496\n",
      "PostgreSQL number of rows:  496\n"
     ]
    }
   ],
   "source": [
    "host = \"localhost\"\n",
    "port = 3306\n",
    "dbname = \"mulesoft\"\n",
    "user = \"root\"\n",
    "password = \"a1b2c3d4!\"\n",
    "\n",
    "while True:\n",
    "\n",
    "    conn_mysql = pymysql.connect(host, user = user,port = port,\n",
    "                               passwd = password, db = dbname)\n",
    "    \n",
    "    engine_postgres = create_engine('postgresql://postgres:a1b2c3d4!@localhost/mulesoft')\n",
    "    \n",
    "    df_mysql = pd.read_sql('select count(*) AS count from Persons;', con = conn_mysql)\n",
    "    \n",
    "    df_postgres = pd.read_sql('select count(*) AS count from Persons;', con = engine_postgres)\n",
    "    conn_mysql.close()\n",
    "    print('MySQL number of rows: ', df_mysql['count'].tolist()[0])\n",
    "    print('PostgreSQL number of rows: ', df_postgres['count'].tolist()[0])\n",
    "    time.sleep(3)"
   ]
  }
 ],
 "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}