{"cells": [{"source": ["#Load the required packages\n", "import numpy as np\n", "import pandas as pd\n", "import multiprocessing as mp\n", "import time\n", "\n", "from subprocess import check_output\n", "import subprocess\n", "files=check_output([\"ls\", \"../input\"]).decode(\"utf8\")\n", "print(files)\n", "#check the number of cores\n", "num_cores = mp.cpu_count()\n", "print(\"This kernel has :\",num_cores,\"cores\")\n", "#Check the number of row of each file\n", "for file in files.split(\"\\n\"):\n", "    path='../input/'+file\n", "    popenobj=subprocess.Popen(['wc', '-l', path], stdout=subprocess.PIPE, \n", "                                              stderr=subprocess.PIPE)\n", "    result,error= popenobj.communicate()\n", "    #print(result,error)\n", "    print(\"The file :\",file,\"has :\",result.strip().split()[0],\"rows\")\n"], "outputs": [], "execution_count": null, "metadata": {"_uuid": "ea24fa0515ce74039f79ad6a70f6add6f89e02fc", "_cell_guid": "85cce2be-733a-4532-9881-8cb657c34d50"}, "cell_type": "code"}, {"source": ["# user logs have 392 million records!!\n", "\n", "The kaggle kernal limitations are as follows:\n", "- It has a self timeout after processing a \"cell\" for ~10 minutes\n", "- Memory limit is 8 GB \n", "Link:https://www.kaggle.com/wiki/Scripts\n", "\n", "Hence by splitting the file into chunks less than 8 GB (I've used a chunk size of 40 mil rows), we can squeeze in the feature extraction with the timeout of 10 minutes.\n"], "metadata": {"_uuid": "d6500e2ab47a68b08f26bcfa8ef2295aede38ecb", "_cell_guid": "f81501cc-2095-4529-b759-e603ae142214"}, "cell_type": "markdown"}, {"source": ["# testing out the funciton for one iteration\n", "import time\n", "size=1e6   # 1 million\n", "user_log_chunks = pd.read_csv('../input/user_logs.csv',chunksize=size,index_col=['msno'])\n", "start_time = time.time()\n", "for i,user_log_chunk in enumerate(user_log_chunks):\n", "    #print(user_log_chunk.head())\n", "    grouped_object=user_log_chunk.groupby(user_log_chunk.index,sort=False) # not sorting results in a minor speedup\n", "    func = {'date':['min','max'], \n", "           'num_25':['sum'],'num_50':['sum'],\n", "           'num_75':['sum'],'num_985':['sum'],\n", "           'num_100':['sum'],'num_unq':['sum'],'total_secs':['sum']}\n", "    if(i==0):  # for the fiest run create the result \n", "        result=grouped_object.agg(func)\n", "        print(len(result))\n", "        print(\"Round \",i,\"took %s seconds\" % (time.time() - start_time))\n", "    else:\n", "        result=result.append(grouped_object.agg(func))\n", "        print(len(result))\n", "        print(\"Round \",i,\"took %s seconds\" % (time.time() - start_time))\n", "        break\n", "    #print(print(\"Round \",i,\"took %s seconds\" % (time.time() - start_time)))\n", "    "], "outputs": [], "execution_count": null, "metadata": {"_kg_hide-output": false, "collapsed": true, "_uuid": "078b25c4d679af19bbbe9d5637e90aa5d83df989", "_cell_guid": "ef8a8fd3-e522-4e44-a8ca-06b61feb76f2", "_kg_hide-input": false}, "cell_type": "code"}, {"source": ["#writing as a function\n", "def process_user_log(chunk):\n", "    grouped_object=chunk.groupby(chunk.index,sort=False) # not sorting results in a minor speedup\n", "    func = {'date':['min','max','count'], \n", "           'num_25':['sum'],'num_50':['sum'],\n", "           'num_75':['sum'],'num_985':['sum'],\n", "           'num_100':['sum'],'num_unq':['sum'],'total_secs':['sum']}\n", "    answer=grouped_object.agg(func)\n", "    return answer\n", "print(\"done\")"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "a8e2c27b35adfaba5f885a5ec54969f72eb48890", "_cell_guid": "4c093ddc-e0cc-48e9-8967-9567765b710e"}, "cell_type": "code"}, {"source": [], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "e85c1dea6fd3bf53278bd27a097c18ccccd83449", "_cell_guid": "9916aa7a-9c18-4763-98c2-fa0fae4af038"}, "cell_type": "code"}, {"source": ["size=4e7 # 40 million\n", "reader = pd.read_csv('../input/user_logs.csv',chunksize=size,index_col=['msno'])\n", "start_time = time.time()\n", "for i in range(10):\n", "    user_log_chunk=next(reader)\n", "    if(i==0):\n", "        result=process_user_log(user_log_chunk)\n", "        print(\"Loop \",i,\"took %s seconds\" % (time.time() - start_time))\n", "    else:\n", "        result=result.append(process_user_log(user_log_chunk))\n", "        print(\"Loop \",i,\"took %s seconds\" % (time.time() - start_time))\n", "    del(user_log_chunk)    \n"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "702fd9f951e15b25c656e7858599bc29d7a2766c", "_cell_guid": "d3d78089-35a7-429c-b40a-e3f6c27cb0da"}, "cell_type": "code"}, {"source": ["  "], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "a413cc82ab730253358161506b6e400eb8966639", "_cell_guid": "d2e89154-8934-4868-a65a-1b9776aff0ec"}, "cell_type": "code"}, {"source": ["print(len(result))\n", "check=result.index.unique()\n", "print(len(check))\n"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "0b7102fd9f144bdbb580de6615ef89ec76b54862", "_cell_guid": "8b447695-dee8-4712-926a-6e52f492b1dd"}, "cell_type": "code"}, {"source": ["#result.columns.values\n", "result.columns = ['_'.join(col).strip() for col in result.columns.values]\n", "result.columns"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "79b5f27a9a484a915ab336b8da90403ad24b8ebc", "_cell_guid": "864f3686-6564-4dc1-bb61-fa736f37b753"}, "cell_type": "code"}, {"source": ["result.head()\n", "func = {'date_min':['min'],'date_max':['max'],'date_count':['count'] ,\n", "           'num_25_sum':['sum'],'num_50_sum':['sum'],\n", "           'num_75_sum':['sum'],'num_985_sum':['sum'],\n", "           'num_100_sum':['sum'],'num_unq_sum':['sum'],'total_secs_sum':['sum']}\n", "processed_user_log=result.groupby(result.index).agg(func)\n", "print(len(processed_user_log))\n", "processed_user_log.columns = processed_user_log.columns.get_level_values(0)\n", "print(processed_user_log.head())"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "e8f6e9f5ddede66a630a113a7dbe676b4688471a", "_cell_guid": "e77526fb-50ae-49ef-a5c3-3ed2586f8ad9"}, "cell_type": "code"}, {"source": ["processed_user_log.head()"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "8bcf1ad269a76bb73c2ff87bdc97b2b4c83aa666", "_cell_guid": "84b73e41-6287-427b-93d7-253c4e69e2e4"}, "cell_type": "code"}, {"source": ["# this function did not work because apparently only lists can be passed back from a function through the \n", "# multiprocessing package \n", "# If we use multiple cores the process can be 16 times faster\n", "\n", "# create as many processes as there are CPUs on your machine\n", "# if __name__ == '__main__':\n", "#     size=1000000\n", "#     parser = pd.read_csv('../input/user_logs.csv',chunksize=size,index_col=['msno'])\n", "#     num_processes = mp.cpu_count()\n", "#     pool=mp.Pool(processes=num_processes)\n", "#     start_time = time.time()\n", "#     list_of_ans=[]\n", "#     def collect_results(result):\n", "#         list_of_ans.extend(result)\n", "#         print(done)\n", "#     for i,df in enumerate(parser):\n", "#         print(\"starting at\",i,\" %s seconds\" % (time.time() - start_time))\n", "#         pool.map(process_user_log,df,callback=collect_results) \n", "#         \n", "#     print(\"action \",i,\"took %s seconds\" % (time.time() - start_time))\n", "#     pool.close()\n", "#     pool.join()\n", "        \n"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "577cd0613941cef1cde92858c3b4ca26e75d201f", "_cell_guid": "fc7321ec-26d5-40cf-a7b3-8bcde9734538"}, "cell_type": "code"}, {"source": ["processed_user_log.to_csv(\"processed_user_log.csv\")"], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "10afc6bdd95f3e9fdc3c03cf02fd3b783fc217f9", "_cell_guid": "7bd991fb-5c9a-45be-8968-0d4c674f7f9f"}, "cell_type": "code"}, {"source": [], "outputs": [], "execution_count": null, "metadata": {"collapsed": true, "_uuid": "fcdc21c8f17c27a6400e35d41d10e72eaf935be2", "_cell_guid": "bb83b44c-d8e9-443c-8cd5-c3ace61ebb9e"}, "cell_type": "code"}], "nbformat_minor": 1, "metadata": {"kernelspec": {"name": "python3", "display_name": "Python 3", "language": "python"}, "language_info": {"name": "python", "nbconvert_exporter": "python", "codemirror_mode": {"version": 3, "name": "ipython"}, "mimetype": "text/x-python", "pygments_lexer": "ipython3", "version": "3.6.3", "file_extension": ".py"}}, "nbformat": 4}