Preprocess_Dask.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:15.203071Z",
     "start_time": "2018-12-22T19:17:13.599738Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['submissions', 'test_set', 'track_features', 'training_set']\n"
     ]
    }
   ],
   "source": [
    "import numpy as np # linear algebra\n",
    "import pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n",
    "pd.set_option('display.max_columns', 500)\n",
    "\n",
    "import dask\n",
    "import dask.dataframe as dd\n",
    "\n",
    "from helpers import *\n",
    "import keggler as kg\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline\n",
    "import seaborn as sns\n",
    "\n",
    "import gc\n",
    "gc.enable()\n",
    "\n",
    "import warnings\n",
    "warnings.simplefilter(action='ignore', category=Warning)\n",
    "\n",
    "import os, psutil\n",
    "\n",
    "# Set up a logger to dump messages to both log file and notebook\n",
    "import logging as logging\n",
    "def ini_log(filename):\n",
    "    logger = logging.getLogger(__name__)\n",
    "    ## avoid multiple printouts due to same handlers added several times\n",
    "    if not logger.handlers:\n",
    "        logger.setLevel(logging.DEBUG)\n",
    "\n",
    "        handlers = [logging.StreamHandler(None), logging.FileHandler(filename, 'a')]\n",
    "\n",
    "        fmt=logging.Formatter('%(asctime)-15s: %(levelname)s  %(message)s')\n",
    "        for h in handlers:\n",
    "            h.setFormatter(fmt)\n",
    "            logger.addHandler(h)\n",
    "    return logger\n",
    "        \n",
    "log = ini_log('out.log')\n",
    "\n",
    "#PATH='data_mini/'\n",
    "#prefix='_mini'\n",
    "\n",
    "PATH='data/'\n",
    "SUB_PATH='test_set/' #test_set #training_set\n",
    "prefix='_0_2018071'\n",
    "\n",
    "print(os.listdir(PATH))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:15.229933Z",
     "start_time": "2018-12-22T19:17:15.205801Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<dask.config.set at 0x7fb313f99048>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from multiprocessing.pool import ThreadPool\n",
    "import dask\n",
    "dask.config.set(pool=ThreadPool(3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-17T11:49:59.083792Z",
     "start_time": "2018-12-17T11:49:59.080943Z"
    }
   },
   "source": [
    "# Read in training data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:15.236374Z",
     "start_time": "2018-12-22T19:17:15.232711Z"
    }
   },
   "outputs": [],
   "source": [
    "# Training\n",
    "# fin = '{}/training_set/log{}*.csv.gz'.format(PATH,prefix)\n",
    "\n",
    "# Test\n",
    "# fin=['{}/{}/log_prehistory_20180715_000000000000.csv.gz'.format(PATH,SUB_PATH),\n",
    "#      '{}/{}/log_prehistory_20180716_000000000000.csv.gz'.format(PATH,SUB_PATH)\n",
    "#     ]\n",
    "fin = '{}/{}/log_input_*.csv.gz'.format(PATH,SUB_PATH)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:15.423650Z",
     "start_time": "2018-12-22T19:17:15.419311Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'data//test_set//log_input_*.csv.gz'"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fin"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:16.191576Z",
     "start_time": "2018-12-22T19:17:16.128113Z"
    }
   },
   "outputs": [],
   "source": [
    "dd_trn = dd.read_csv(fin,\n",
    "                     #usecols=[i for i in range(21) if i != 16], # disable this to process data/test_set/log_input_\n",
    "                     compression='gzip')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:17.030658Z",
     "start_time": "2018-12-22T19:17:17.028242Z"
    }
   },
   "outputs": [],
   "source": [
    "# dd_trn.memory_usage(deep=True).compute() / 1024**2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:17.860399Z",
     "start_time": "2018-12-22T19:17:17.855917Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "session_id          object\n",
       "track_id_clean      object\n",
       "session_position     int64\n",
       "session_length       int64\n",
       "dtype: object"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dd_trn.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:19.965293Z",
     "start_time": "2018-12-22T19:17:19.962797Z"
    },
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "#dd_trn.head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reduce memory footprint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:22.197005Z",
     "start_time": "2018-12-22T19:17:22.193040Z"
    }
   },
   "outputs": [],
   "source": [
    "reasons = {'trackdone': 1, 'fwdbtn': 2, 'trackerror': 8, 'remote': 7, 'clickrow': 4, 'backbtn': 3, 'playbtn': 6, 'appload': 5, 'endplay': 9, 'logout':10}\n",
    "enc = {\n",
    "    'hist_user_behavior_reason_start': reasons,\n",
    "    'context_type': {'radio': 3, 'personalized_playlist': 4, 'charts': 6, 'user_collection': 2, 'editorial_playlist': 1, 'catalog': 5},\n",
    "    'hist_user_behavior_reason_end': reasons\n",
    "}\n",
    "\n",
    "for c in enc:\n",
    "    if c in dd_trn.columns:\n",
    "        dd_trn[c] = dd_trn[c].map(enc[c]).fillna(0).astype(np.uint8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:23.016178Z",
     "start_time": "2018-12-22T19:17:23.013775Z"
    }
   },
   "outputs": [],
   "source": [
    "# dd_trn.memory_usage(deep=True).sum().compute()/1024**2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:23.714477Z",
     "start_time": "2018-12-22T19:17:23.711476Z"
    }
   },
   "outputs": [],
   "source": [
    "# dd_trn.memory_usage(deep=True).compute()/1024**2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:17:24.623928Z",
     "start_time": "2018-12-22T19:17:24.619299Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "session_id          object\n",
       "track_id_clean      object\n",
       "session_position     int64\n",
       "session_length       int64\n",
       "dtype: object"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dd_trn.dtypes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-17T11:03:14.866890Z",
     "start_time": "2018-12-17T11:03:14.861479Z"
    }
   },
   "source": [
    "# Read in track features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:18:16.308969Z",
     "start_time": "2018-12-22T19:17:27.337585Z"
    }
   },
   "outputs": [],
   "source": [
    "df_trk = pd.concat([pd.read_csv(PATH+'track_features/'+f, \n",
    "                     usecols=range(22), \n",
    "                     dtype={'release_year': np.uint32},\n",
    "                     compression='gzip'\n",
    "                    ) for f in os.listdir(PATH+'track_features/') if f.startswith('tf_')\n",
    "         ], axis=0).set_index('track_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:18:16.553518Z",
     "start_time": "2018-12-22T19:18:16.310667Z"
    },
    "hide_input": false
   },
   "outputs": [],
   "source": [
    "df_trk['mode'] = df_trk['mode'].map({'major': 0, 'minor':1}).astype(np.uint8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:18:16.852677Z",
     "start_time": "2018-12-22T19:18:16.555624Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Index                     335.795269\n",
       "duration                   28.277496\n",
       "release_year               14.138748\n",
       "us_popularity_estimate     28.277496\n",
       "acousticness               28.277496\n",
       "beat_strength              28.277496\n",
       "bounciness                 28.277496\n",
       "danceability               28.277496\n",
       "dyn_range_mean             28.277496\n",
       "energy                     28.277496\n",
       "flatness                   28.277496\n",
       "instrumentalness           28.277496\n",
       "key                        28.277496\n",
       "liveness                   28.277496\n",
       "loudness                   28.277496\n",
       "mechanism                  28.277496\n",
       "mode                        3.534687\n",
       "organism                   28.277496\n",
       "speechiness                28.277496\n",
       "tempo                      28.277496\n",
       "time_signature             28.277496\n",
       "valence                    28.277496\n",
       "dtype: float64"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_trk.memory_usage(deep=True)/1024**2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-17T11:53:39.360659Z",
     "start_time": "2018-12-17T11:53:39.355981Z"
    }
   },
   "source": [
    "# Merge DS with Track features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:18:28.699724Z",
     "start_time": "2018-12-22T19:18:16.854923Z"
    }
   },
   "outputs": [],
   "source": [
    "dd_out = dd_trn.merge(df_trk, \n",
    "                      how='left', \n",
    "                      right_index=True, \n",
    "                      left_on='track_id_clean'\n",
    "                     ).drop(['track_id_clean'], \n",
    "                            axis=1\n",
    "                           )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cast boolean into `unit8` to reduce file size"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:20:51.496208Z",
     "start_time": "2018-12-22T19:20:51.413622Z"
    }
   },
   "outputs": [],
   "source": [
    "for c,dtype in col_dtype.items():\n",
    "    if c in dd_out.columns:\n",
    "        dd_out[c] = dd_out[c].astype(dtype)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:20:52.406217Z",
     "start_time": "2018-12-22T19:20:52.400684Z"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "session_id                 object\n",
       "session_position             int8\n",
       "session_length               int8\n",
       "duration                  float16\n",
       "release_year                int16\n",
       "us_popularity_estimate    float16\n",
       "acousticness              float16\n",
       "beat_strength             float16\n",
       "bounciness                float16\n",
       "danceability              float16\n",
       "dyn_range_mean            float16\n",
       "energy                    float16\n",
       "flatness                  float16\n",
       "instrumentalness          float16\n",
       "key                         int16\n",
       "liveness                  float16\n",
       "loudness                  float16\n",
       "mechanism                 float16\n",
       "mode                         int8\n",
       "organism                  float16\n",
       "speechiness               float16\n",
       "tempo                     float16\n",
       "time_signature              int16\n",
       "valence                   float16\n",
       "dtype: object"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dd_out.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:20:53.299513Z",
     "start_time": "2018-12-22T19:20:53.296393Z"
    },
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "# dd_out.memory_usage(deep=True).sum().compute()/1024**2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Store the DD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:20:55.891065Z",
     "start_time": "2018-12-22T19:20:55.887223Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "66"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dd_out.npartitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-12-22T19:43:36.541566Z",
     "start_time": "2018-12-22T19:20:57.800622Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['data//test_set//subDD_v2_00.h5',\n",
       " 'data//test_set//subDD_v2_01.h5',\n",
       " 'data//test_set//subDD_v2_02.h5',\n",
       " 'data//test_set//subDD_v2_03.h5',\n",
       " 'data//test_set//subDD_v2_04.h5',\n",
       " 'data//test_set//subDD_v2_05.h5',\n",
       " 'data//test_set//subDD_v2_06.h5',\n",
       " 'data//test_set//subDD_v2_07.h5',\n",
       " 'data//test_set//subDD_v2_08.h5',\n",
       " 'data//test_set//subDD_v2_09.h5',\n",
       " 'data//test_set//subDD_v2_10.h5',\n",
       " 'data//test_set//subDD_v2_11.h5',\n",
       " 'data//test_set//subDD_v2_12.h5',\n",
       " 'data//test_set//subDD_v2_13.h5',\n",
       " 'data//test_set//subDD_v2_14.h5',\n",
       " 'data//test_set//subDD_v2_15.h5',\n",
       " 'data//test_set//subDD_v2_16.h5',\n",
       " 'data//test_set//subDD_v2_17.h5',\n",
       " 'data//test_set//subDD_v2_18.h5',\n",
       " 'data//test_set//subDD_v2_19.h5',\n",
       " 'data//test_set//subDD_v2_20.h5',\n",
       " 'data//test_set//subDD_v2_21.h5',\n",
       " 'data//test_set//subDD_v2_22.h5',\n",
       " 'data//test_set//subDD_v2_23.h5',\n",
       " 'data//test_set//subDD_v2_24.h5',\n",
       " 'data//test_set//subDD_v2_25.h5',\n",
       " 'data//test_set//subDD_v2_26.h5',\n",
       " 'data//test_set//subDD_v2_27.h5',\n",
       " 'data//test_set//subDD_v2_28.h5',\n",
       " 'data//test_set//subDD_v2_29.h5',\n",
       " 'data//test_set//subDD_v2_30.h5',\n",
       " 'data//test_set//subDD_v2_31.h5',\n",
       " 'data//test_set//subDD_v2_32.h5',\n",
       " 'data//test_set//subDD_v2_33.h5',\n",
       " 'data//test_set//subDD_v2_34.h5',\n",
       " 'data//test_set//subDD_v2_35.h5',\n",
       " 'data//test_set//subDD_v2_36.h5',\n",
       " 'data//test_set//subDD_v2_37.h5',\n",
       " 'data//test_set//subDD_v2_38.h5',\n",
       " 'data//test_set//subDD_v2_39.h5',\n",
       " 'data//test_set//subDD_v2_40.h5',\n",
       " 'data//test_set//subDD_v2_41.h5',\n",
       " 'data//test_set//subDD_v2_42.h5',\n",
       " 'data//test_set//subDD_v2_43.h5',\n",
       " 'data//test_set//subDD_v2_44.h5',\n",
       " 'data//test_set//subDD_v2_45.h5',\n",
       " 'data//test_set//subDD_v2_46.h5',\n",
       " 'data//test_set//subDD_v2_47.h5',\n",
       " 'data//test_set//subDD_v2_48.h5',\n",
       " 'data//test_set//subDD_v2_49.h5',\n",
       " 'data//test_set//subDD_v2_50.h5',\n",
       " 'data//test_set//subDD_v2_51.h5',\n",
       " 'data//test_set//subDD_v2_52.h5',\n",
       " 'data//test_set//subDD_v2_53.h5',\n",
       " 'data//test_set//subDD_v2_54.h5',\n",
       " 'data//test_set//subDD_v2_55.h5',\n",
       " 'data//test_set//subDD_v2_56.h5',\n",
       " 'data//test_set//subDD_v2_57.h5',\n",
       " 'data//test_set//subDD_v2_58.h5',\n",
       " 'data//test_set//subDD_v2_59.h5',\n",
       " 'data//test_set//subDD_v2_60.h5',\n",
       " 'data//test_set//subDD_v2_61.h5',\n",
       " 'data//test_set//subDD_v2_62.h5',\n",
       " 'data//test_set//subDD_v2_63.h5',\n",
       " 'data//test_set//subDD_v2_64.h5',\n",
       " 'data//test_set//subDD_v2_65.h5']"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# dd_out.to_csv(PATH+'/'+SUB_PATH+'/outDD_v2_*.csv.gz', index=False, float_format='%.5f', compression='gzip')\n",
    "dd_out.to_hdf(PATH+'/'+SUB_PATH+'/subDD_v2_*.h5', key='df')"
   ]
  },
  {
   "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.5.5"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}