examples-sql.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Engineering in Python with databolt  - Fast Loading to SQL with pandas (d6tlib/d6tstack)\n",
    "\n",
    "Pandas and SQL are great but they have some problems:\n",
    "* loading data from pandas to SQL is very slow. So you can't preprocess data with python and then quickly store it in a db\n",
    "* Loading CSV files into SQL is cumbersome and quickly breaks when input files are not consistent\n",
    "\n",
    "With `d6tstack` you can:\n",
    "* load pandas dataframes to postgres or mysql much faster than with `pd.to_sql()` and with minimal memory consumption\n",
    "* preprocess CSV files with pandas before writing to db\n",
    "* solve data schema problems (eg new or renamed columns) before writing to db \n",
    "* out of core functionality where large files are processed in chunks\n",
    "\n",
    "In this workbook we will demonstrate the usage of the d6tstack library for quickly loading data into SQL from CSV files and pandas."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# pd.to_sql() is slow\n",
    "Let's see how slow `pd.to_sql()` is storing 100k rows of random data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(100000, 23)\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import uuid\n",
    "import sqlalchemy\n",
    "import glob\n",
    "import time\n",
    "\n",
    "cfg_uri_psql = 'postgresql+psycopg2://psqlusr:psqlpwdpsqlpwd@localhost/psqltest'\n",
    "cfg_uri_mysql = 'mysql+mysqlconnector://testusr:testpwd@localhost/testdb'\n",
    "\n",
    "cfg_nobs = int(1e5)\n",
    "np.random.seed(0)\n",
    "df = pd.DataFrame({'id':range(cfg_nobs)})\n",
    "df['uuid']=[uuid.uuid4().hex.upper()[0:10] for _ in range(cfg_nobs)]\n",
    "df['date']=pd.date_range('1/1/2010',periods=cfg_nobs, freq='1T')\n",
    "for i in range(20):\n",
    "    df['d'+str(i)]=np.random.normal(size=int(cfg_nobs))\n",
    "\n",
    "print(df.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--- 28.010449647903442 seconds ---\n"
     ]
    }
   ],
   "source": [
    "sqlengine = sqlalchemy.create_engine(cfg_uri_psql)\n",
    "\n",
    "start_time = time.time()\n",
    "df.to_sql('benchmark',sqlengine,if_exists='replace')\n",
    "print(\"--- %s seconds ---\" % (time.time() - start_time))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Speeding up pd.to_sql() in postgres and mysql with d6tstack\n",
    "Let's see how we can make this faster. In this simple example we have a ~5x speedup with the speedup growing exponentially with larger datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--- 4.6529316902160645 seconds ---\n",
      "creating mysql.csv ok\n",
      "loading mysql.csv ok\n",
      "--- 7.102342367172241 seconds ---\n"
     ]
    }
   ],
   "source": [
    "import d6tstack.utils\n",
    "\n",
    "# psql\n",
    "start_time = time.time()\n",
    "d6tstack.utils.pd_to_psql(df, cfg_uri_psql, 'benchmark', if_exists='replace')\n",
    "print(\"--- %s seconds ---\" % (time.time() - start_time))\n",
    "\n",
    "# mysql\n",
    "start_time = time.time()\n",
    "d6tstack.utils.pd_to_mysql(df, cfg_uri_mysql, 'benchmark', if_exists='replace')\n",
    "print(\"--- %s seconds ---\" % (time.time() - start_time))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using Pandas for preprocessing CSVs before storing to database\n",
    "Pandas is great for preprocessing data. For example lets say we want to process dates before importing them to a database. `d6tstack` makes this easy for you, you simply pass the filename or list of files along with the preprocessing function and it will be quickly loaded in SQL - without loading everything into memory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "         date  sales  cost  profit\n",
      "0  2011-02-01    200   -90     110\n",
      "1  2011-02-02    200   -90     110\n",
      "2  2011-02-03    200   -90     110\n",
      "3  2011-02-04    200   -90     110\n",
      "4  2011-02-05    200   -90     110\n"
     ]
    }
   ],
   "source": [
    "cfg_fname = 'test-data/input/test-data-input-csv-colmismatch-feb.csv'\n",
    "print(pd.read_csv(cfg_fname).head())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sniffing columns ok\n",
      "        date  sales  cost  profit date_year_quarter date_monthend\n",
      "0 2011-02-01    200   -90     110              11Q1    2011-02-28\n",
      "1 2011-02-02    200   -90     110              11Q1    2011-02-28\n",
      "2 2011-02-03    200   -90     110              11Q1    2011-02-28\n",
      "3 2011-02-04    200   -90     110              11Q1    2011-02-28\n",
      "4 2011-02-05    200   -90     110              11Q1    2011-02-28\n"
     ]
    }
   ],
   "source": [
    "def apply(dfg):\n",
    "    dfg['date'] = pd.to_datetime(dfg['date'], format='%Y-%m-%d')\n",
    "    dfg['date_year_quarter'] = (dfg['date'].dt.year).astype(str).str[-2:]+'Q'+(dfg['date'].dt.quarter).astype(str)\n",
    "    dfg['date_monthend'] = dfg['date'] + pd.tseries.offsets.MonthEnd()\n",
    "    return dfg\n",
    "\n",
    "d6tstack.combine_csv.CombinerCSV([cfg_fname], apply_after_read=apply,add_filename=False).to_psql_combine(cfg_uri_psql, 'benchmark', if_exists='replace')\n",
    "print(pd.read_sql_table('benchmark',sqlengine).head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Loading multiple CSV to SQL with data schema changes\n",
    "Native database import commands only support one file. You can write a script to process multipe files which first of all is annoying and even worse it often breaks eg if there are schema changes. With `d6tstack` you quickly import multiple files and deal with data schema changes with just a couple of lines of python. The below is a quick example, to explore full functionality see  https://github.com/d6t/d6tstack/blob/master/examples-csv.ipynb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sniffing columns ok\n",
      "all equal False\n",
      "\n"
     ]
    },
    {
     "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>date</th>\n",
       "      <th>sales</th>\n",
       "      <th>cost</th>\n",
       "      <th>profit</th>\n",
       "      <th>profit2</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>file_path</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>test-data/input/test-data-input-csv-colmismatch-feb.csv</th>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>test-data/input/test-data-input-csv-colmismatch-jan.csv</th>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>test-data/input/test-data-input-csv-colmismatch-mar.csv</th>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "      <td>True</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                    date  sales  cost  profit  profit2\n",
       "file_path                                                                             \n",
       "test-data/input/test-data-input-csv-colmismatch...  True   True  True    True    False\n",
       "test-data/input/test-data-input-csv-colmismatch...  True   True  True    True    False\n",
       "test-data/input/test-data-input-csv-colmismatch...  True   True  True    True     True"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import glob\n",
    "import d6tstack.combine_csv\n",
    "\n",
    "cfg_fnames = list(glob.glob('test-data/input/test-data-input-csv-colmismatch-*.csv'))\n",
    "c = d6tstack.combine_csv.CombinerCSV(cfg_fnames)\n",
    "\n",
    "# check columns\n",
    "print('all equal',c.is_all_equal())\n",
    "print('')\n",
    "c.is_column_present()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The presence of the additional `profit2` column in the 3rd file would break the data load. `d6tstack` will fix the situation and load everything correctly. And you can run any additional preprocessing logic like in the above example. All this is done out of core so you can process even large files without any memory issues."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sniffing columns ok\n",
      "         date  sales  cost  profit  profit2 date_year_quarter date_monthend\n",
      "25 2011-03-06    300  -100     200    400.0              11Q1    2011-03-31\n",
      "26 2011-03-07    300  -100     200    400.0              11Q1    2011-03-31\n",
      "27 2011-03-08    300  -100     200    400.0              11Q1    2011-03-31\n",
      "28 2011-03-09    300  -100     200    400.0              11Q1    2011-03-31\n",
      "29 2011-03-10    300  -100     200    400.0              11Q1    2011-03-31\n"
     ]
    }
   ],
   "source": [
    "cfg_fnames = list(glob.glob('test-data/input/test-data-input-csv-colmismatch-*.csv'))\n",
    "d6tstack.combine_csv.CombinerCSV(cfg_fnames, apply_after_read=apply,add_filename=False).to_psql_combine(cfg_uri_psql, 'benchmark', if_exists='replace')\n",
    "print(pd.read_sql_table('benchmark',sqlengine).tail())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}