Past_label_extraction.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Label extraction\n",
    "Leer los files crudos de geo_voice data\n",
    "\n",
    "Encontrar si un user es epidemico u expuesto en los meses previos al dataset donde se esta analizando. Estas serian las variables que querriamos predecir.\n",
    "\n",
    "La idea es crear la lista que cuenta antennas utilizadas para cada user en el dataset outputeado por simple_format diariamente y agrupar esta lista (con repeticiones) en un agregado mensual (para luego determinar home_antenna etc.)\n",
    "\n",
    "Tambien usando info de las antenas (epidemicas o no). Podriamos ver si un user \"vive\" o \"paso\" por la zona epidemica.\n",
    "\n",
    "## definiciones utilizadas\n",
    "* timestamp arranca en 0 segundos para 01/01/2012 00:00am\n",
    "* el analisis es solo para cdrs con usuarios de la TelCo. Los users de la TelCo van todos en la primer columna (independientemente del nombre de la columna..) entonces si la llamada es entrante quiere decir que el user de la telco esta recibiendo un llamado en esa antennaID. si la llamada es saliente es al reves. (Podria pasar que sea comunicaciones inter-Telco y ahi habria dos records uno saliente y otro entrante pero con los mismos 2 userIDs).\n",
    "* La Direction, viene dada relativa al user de la primer columna (incoming si entra y viceversa si es outgoing).\n",
    "\n",
    "### el dataset de simple_format vendria con este header\n",
    " {'USER':np.int32 \\\\\\  'OTHER_USER':np.uint32 \\\\\\ 'Direction':np.object \\\\\\  'TimeStamp':np.uint32  \\\\\\ 'Duration':np.uint16 \\\\\\  'AntennaID':np.uint16}\n",
    "\n",
    "por lo cual seria conveniente asignar dtypes para cargar mas rapido en memoria"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "application/javascript": [
       "IPython.notebook.set_autosave_interval(180000)"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Autosaving every 180 seconds\n"
     ]
    }
   ],
   "source": [
    "%autosave 180"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pandas as pd; \n",
    "import numpy as np; \n",
    "import os;\n",
    "import random;\n",
    "pd.set_option('display.max_rows', 300)\n",
    "from matplotlib import pyplot as plt\n",
    "%matplotlib inline\n",
    "import time\n",
    "np.random.seed(2016)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['SimpleFormat',\n",
       " 'simple_format.sh',\n",
       " 'simple_format_201401.txt.gz',\n",
       " 'simple_format_201402.txt.gz',\n",
       " 'simple_format_201403.txt.gz',\n",
       " 'simple_format_201404.txt.gz',\n",
       " 'simple_format_201405.txt.gz',\n",
       " 'simple_format_201406.txt.gz',\n",
       " 'simple_format_201407.txt.gz',\n",
       " 'simple_format_201408.txt.gz',\n",
       " 'simple_format_201409.txt.gz',\n",
       " 'simple_format_201410.txt.gz',\n",
       " 'simple_format_201411.txt.gz',\n",
       " 'simple_format_201412.txt.gz',\n",
       " 'simple_format_201501.txt.gz',\n",
       " 'simple_format_201501_head.txt.gz',\n",
       " 'simple_format_201502.txt.gz',\n",
       " 'simple_format_201503.txt.gz',\n",
       " 'simple_format_201504.txt.gz',\n",
       " 'simple_format_201505.txt.gz',\n",
       " 'simple_format_201506.txt.gz',\n",
       " 'simple_format_201507.txt.gz',\n",
       " 'simple_format_201508.txt.gz',\n",
       " 'simple_format_201509.txt.gz',\n",
       " 'surrogated_antennas.txt.gz',\n",
       " 'surrogated_antennas_copy.txt.gz',\n",
       " 'surrogated_nodes.txt.gz']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "!!ls /grandata/simple_format/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\r\n"
     ]
    }
   ],
   "source": [
    "!zcat /grandata/simple_format/simple_format_201404.txt.gz | head "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#ejemplo  \"/grandata/simple_format/01-02012014.txt.gz\"\n",
    "#seteamos el lugar de trabajo\n",
    "rootdir=\"/grandata/simple_format/\"\n",
    "os.chdir(rootdir)\n",
    "year = \"2014\"; \n",
    "month_start= \"1\";\n",
    "if int(month_start )<10:\n",
    "    month_start=\"0\"+month_start\n",
    "    \n",
    "day_start = \"01\"\n",
    "day_end = str(int(day_start)+1)\n",
    "if int(day_end )<10:\n",
    "    day_end=\"0\"+ day_end\n",
    "\n",
    "input_file= rootdir +\"simple_format_{y}{ms}.txt.gz\"\\\n",
    "                .format(y=year,ms=month_start,\n",
    "                       ds=day_start,de=day_end)\n",
    "    \n",
    "def get_output_file(night_filter=False,week_end=False):\n",
    "    output = \"/home/juan/mobility-study/past_labels\"\n",
    "    if week_end == True:\n",
    "        output = output + \"_wkend\"\n",
    "    if night_filter == True:\n",
    "        output = output + \"_ngtfilter\"\n",
    "    return output + \".txt.gz\"\n",
    "\n",
    "output_file = get_output_file()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#para enriquecer el dataset de CDRs con info acerca de la zona epidemica etc.\n",
    "antennas= pd.read_csv('/home/juan/mobility-study/antennas_mexico.csv',sep = \"|\",header=0,index_col=0)\n",
    "antennas.index.rename(\"ANTENNA_ID\",inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "LineKeyOrigin|CallsWeekDaylight|CallsWeekDaylight_EPI|CallsWeekNight|CallsWeekNight_EPI|CallsWeekend|CallsWeekend_EPI|TimeWeekDaylight|TimeWeekDaylight_EPI|TimeWeekNight|TimeWeekNight_EPI|TimeWeekend|TimeWeekend_EPI|TOTAL_USERS|EPI_USERS|EXP_USERS\r\n",
      "000004F96010462FA7A71C7784E27476|0|0|0|0|0|1|0|0|0|0|0|34|1|0|1\r\n",
      "000007EEE7895E17937CFF86379C8E17|0|2|0|0|0|0|0|144|0|0|0|0|1|1|1\r\n",
      "00000837B868F926F67BD440A0845C1D|1|0|2|4|3|0|47|0|74|86|109|0|3|0|2\r\n",
      "00000C5D3BFBEBD919F6F68386C3DD5E|4|0|0|0|0|0|281|0|0|0|0|0|1|0|1\r\n",
      "0000129F4191FFD0963774D2307C7510|2|0|0|0|0|0|66|0|0|0|0|0|2|0|2\r\n",
      "000017368A8964F8C71A9F3A1EAF6160|7|4|0|0|1|0|960|630|0|0|58|0|4|1|4\r\n",
      "00001ADC8C5705AB4FD8197B5024C1AA|7|1|5|5|4|1|744|29|207|335|218|141|4|2|4\r\n",
      "00001B448AE41076EB2C58ABEAF0E0C8|15|0|12|1|10|1|1319|0|705|31|459|29|2|1|2\r\n",
      "00001C544BEBBB9109CE3552B4CDC490|0|0|1|0|0|0|0|0|36|0|0|0|1|0|1\r\n",
      "\r\n",
      "gzip: stdout: Broken pipe\r\n"
     ]
    }
   ],
   "source": [
    "user_hashes_file= \"/home/juan/mobility-study/output_sum_links.txt.gz\"\n",
    "!zcat $user_hashes_file | head"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#levanto la tabla de users hash_map.\n",
    "user_hashes = pd.read_csv(\"/home/juan/mobility-study/output_sum_links.txt.gz\",\n",
    "                                  sep=\"|\",\n",
    "                                  header=0,\n",
    "        dtype = {'LineKeyOrigin':np.object_,'CallsWeekDaylight':np.uint16,'CallsWeekDaylight_EPI':np.uint16,'CallsWeekNight':np.uint16,\n",
    "            'CallsWeekNight_EPI':np.uint16,'CallsWeekend':np.uint16,'CallsWeekend_EPI':np.uint16, 'TimeWeekDaylight':np.uint16,\n",
    "                'TimeWeekDaylight_EPI':np.uint16,'TimeWeekNight':np.uint16,'TimeWeekNight_EPI':np.uint16,'TimeWeekEnd':np.uint16\n",
    "                ,'TimeWeekEnd_EPI':np.uint16,'TOTAL_USERS':np.uint16,'EPI_USERS':np.uint16,'EXP_USERS':np.uint16},\n",
    "                          usecols = ['LineKeyOrigin'],\n",
    "                )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "week_end = False\n",
    "night_filter = False\n",
    "\n",
    "output_file = get_output_file()\n",
    "sample_file = output_file.replace(\".txt.gz\",\"_sample.txt.gz\")\n",
    "#para el manejo de las distintas salidas del programa"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Corremos test sobre un sample only"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "428361977\r\n"
     ]
    }
   ],
   "source": [
    "!zcat $input_file | wc -l"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#leo algunas primeras lineas\n",
    "sample = pd.read_csv(\n",
    "                input_file,\n",
    "                engine = 'c',\n",
    "#                chunksize = 10**7,\n",
    "    #            iterator =True,\n",
    "                nrows = 3*10**7,\n",
    "                sep = ' ',\n",
    "                header = 0,\n",
    "                index_col=None,\n",
    "                usecols = ['USER','ANTENNA_ID'],\n",
    "                dtype = {'TIMESTAMP':np.uint32,'ANTENNA_ID':np.uint16,'USER':np.object_,'OTHER_USER':np.object_}\n",
    "                )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(2449242, (2449242, 1), (30000000, 4))"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(np.unique(user_hashes.values)),user_hashes.shape,sample.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>LATITUDE</th>\n",
       "      <th>LONGITUDE</th>\n",
       "      <th>STATE</th>\n",
       "      <th>EPIDEMIC</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>ANTENNA_ID</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2797</th>\n",
       "      <td>14.671</td>\n",
       "      <td>-92.372</td>\n",
       "      <td>Chiapas</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3060</th>\n",
       "      <td>14.682</td>\n",
       "      <td>-92.155</td>\n",
       "      <td>Chiapas</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3344</th>\n",
       "      <td>14.721</td>\n",
       "      <td>-92.424</td>\n",
       "      <td>Chiapas</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3487</th>\n",
       "      <td>14.778</td>\n",
       "      <td>-92.179</td>\n",
       "      <td>Chiapas</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1924</th>\n",
       "      <td>14.786</td>\n",
       "      <td>-92.366</td>\n",
       "      <td>Chiapas</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "            LATITUDE  LONGITUDE    STATE  EPIDEMIC\n",
       "ANTENNA_ID                                        \n",
       "2797          14.671    -92.372  Chiapas         1\n",
       "3060          14.682    -92.155  Chiapas         1\n",
       "3344          14.721    -92.424  Chiapas         1\n",
       "3487          14.778    -92.179  Chiapas         1\n",
       "1924          14.786    -92.366  Chiapas         1"
      ]
     },
     "execution_count": 75,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "antennas.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sample = sample[sample['USER'].isin(user_hashes.loc[:,'LineKeyOrigin'].values)]\n",
    "# Obs. no es neceario filtrar por usuarios que tengan un rango de llamados minimos/maximos mensuales pues eso ya \n",
    "# estuvo previamente hecho en el filtrado anterior de sum_links y simple_format\n",
    "\n",
    "grouped = sample.groupby(['USER', 'ANTENNA_ID'])['ANTENNA_ID'].agg({'count': np.size})\n",
    "grouped.reset_index(inplace=True,drop=False)\n",
    "\n",
    "#reordeno dentro de c/ USER por el count del antenna, esto me sirve para despues ordenar las antenas por uso\n",
    "grouped.sort_values(by=['USER','count'],ascending=False,inplace=True)\n",
    "grouped = pd.merge(grouped,antennas['EPIDEMIC'].reset_index(), on='ANTENNA_ID',how = 'left')\n",
    "\n",
    "##enriquezco la muestra con datos epidemicos\n",
    "#primero agrego a cada antenna del df el dato de si es epidemica\n",
    "#despues agrupo por el USER y me fijo solo la columna epidemica en c/grupo\n",
    "#finalmente sumo en c/ grupo y tomo la parte superior \n",
    "#entera de esa division con el largo del grupo. Si uso al menos una antena epidemica entonces esta expuesto(==1) Si no,\n",
    "# da 0 pues no estuvo expuesto.        \n",
    "##enriquezco la muestra con datos epidemicos\n",
    "\n",
    "exposed_info = grouped.groupby('USER')['EPIDEMIC'].\\\n",
    "    agg({'EXPOSED' : lambda x: int( np.ceil(np.sum(x)*1.0/np.size(x)) )})\n",
    "\n",
    "#actualizo la tabla grouped\n",
    "grouped = grouped.join(exposed_info['EXPOSED'],on=\"USER\",how=\"left\")\n",
    "\n",
    "del exposed_info\n",
    "\n",
    "output_table = grouped.drop_duplicates(subset = 'USER', keep='first').copy()\n",
    "output_table.drop('count',axis=1,inplace=True)\n",
    "output_table = output_table.set_index('USER')\n",
    "# y aca vamos al write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Main algorithm "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ojo con los s_f corruptos\n",
    "el 201404"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "7.5G\t/grandata/simple_format/simple_format_201401.txt.gz\r\n",
      "6.8G\t/grandata/simple_format/simple_format_201402.txt.gz\r\n",
      "7.4G\t/grandata/simple_format/simple_format_201403.txt.gz\r\n",
      "4.0K\t/grandata/simple_format/simple_format_201404.txt.gz\r\n",
      "7.9G\t/grandata/simple_format/simple_format_201405.txt.gz\r\n",
      "7.5G\t/grandata/simple_format/simple_format_201406.txt.gz\r\n",
      "7.7G\t/grandata/simple_format/simple_format_201407.txt.gz\r\n",
      "7.8G\t/grandata/simple_format/simple_format_201408.txt.gz\r\n",
      "6.1G\t/grandata/simple_format/simple_format_201409.txt.gz\r\n",
      "6.1G\t/grandata/simple_format/simple_format_201410.txt.gz\r\n",
      "6.3G\t/grandata/simple_format/simple_format_201411.txt.gz\r\n",
      "7.0G\t/grandata/simple_format/simple_format_201412.txt.gz\r\n",
      "6.6G\t/grandata/simple_format/simple_format_201501.txt.gz\r\n",
      "4.0K\t/grandata/simple_format/simple_format_201501_head.txt.gz\r\n",
      "6.5G\t/grandata/simple_format/simple_format_201502.txt.gz\r\n",
      "7.7G\t/grandata/simple_format/simple_format_201503.txt.gz\r\n",
      "7.2G\t/grandata/simple_format/simple_format_201504.txt.gz\r\n",
      "19G\t/grandata/simple_format/simple_format_201505.txt.gz\r\n",
      "15G\t/grandata/simple_format/simple_format_201506.txt.gz\r\n",
      "14G\t/grandata/simple_format/simple_format_201507.txt.gz\r\n",
      "15G\t/grandata/simple_format/simple_format_201508.txt.gz\r\n",
      "15G\t/grandata/simple_format/simple_format_201509.txt.gz\r\n"
     ]
    }
   ],
   "source": [
    "!du -ha /grandata/simple_format/simple_format_* "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\r\n"
     ]
    }
   ],
   "source": [
    "!zcat /grandata/simple_format/simple_format_201404.txt.gz | wc -l"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4.0K\t/grandata/simple_format/simple_format_201404.txt.gz\r\n"
     ]
    }
   ],
   "source": [
    "!du -ha /grandata/simple_format/simple_format_201404.txt.gz "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\r\n"
     ]
    }
   ],
   "source": [
    "!zcat /grandata/simple_format/simple_format_201404.txt.gz | head"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Just checking..\n",
    "Vemos consistencia de archivos "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "201401\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 BA694194F2CF7978466AA9602DD38F36 O 63161302 56 1\n",
      "216E0E920C0B51093276E270CC36CB72 BA694194F2CF7978466AA9602DD38F36 O 63162373 88 1\n",
      "216E0E920C0B51093276E270CC36CB72 BCE827A6E83B59F75031D4FCEC36B7DD I 63217965 600 2\n",
      "216E0E920C0B51093276E270CC36CB72 BCE827A6E83B59F75031D4FCEC36B7DD I 63218508 63 2\n",
      "216E0E920C0B51093276E270CC36CB72 BCE827A6E83B59F75031D4FCEC36B7DD I 63217516 93 2\n",
      "216E0E920C0B51093276E270CC36CB72 EADDE544AB1EC4B975AE41337ACFB5B5 I 63228387 35 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 39ABF1E2D91107D64EE4B6719C2D6C60 O 63286717 342 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 995362BDBB319FB4E367EB52E2540CC7 O 63280397 33 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 9D0AFC1877F147C63E235CF25B42CC0B O 63307690 38 4\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201402\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 72868AF4DE00F3FA16B151849D0243F1 I 65891075 150 1\n",
      "216E0E920C0B51093276E270CC36CB72 9D0AFC1877F147C63E235CF25B42CC0B O 65841554 3 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 240891E8E956EFD2E2B5F5603D28BF73 I 65907296 59 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 39ABF1E2D91107D64EE4B6719C2D6C60 O 65908819 35 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 39ABF1E2D91107D64EE4B6719C2D6C60 O 66003608 57 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 DC7CC3B2BA3A97C61582CFB8FD6F4421 I 65975909 58 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 DC7CC3B2BA3A97C61582CFB8FD6F4421 O 66003489 20 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 DC7CC3B2BA3A97C61582CFB8FD6F4421 O 66003672 50 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 F2EE0D5CCA5803309203154CBB9FD221 I 65889851 24 5\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201403\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 7024556BC1620716E79D112100C2DBCA I 68323194 35 1\n",
      "216E0E920C0B51093276E270CC36CB72 EADDE544AB1EC4B975AE41337ACFB5B5 I 68378938 14 2\n",
      "216E0E920C0B51093276E270CC36CB72 EADDE544AB1EC4B975AE41337ACFB5B5 I 68292803 17 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 20D79022A682363CF9227800323C8DFE I 68313107 78 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 39ABF1E2D91107D64EE4B6719C2D6C60 I 68322748 127 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 39ABF1E2D91107D64EE4B6719C2D6C60 O 68313288 132 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 9460EB3EFD3F56DE83970BF1A090854F O 68296635 160 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 9D0AFC1877F147C63E235CF25B42CC0B O 68313382 71 5\n",
      "73C2C1F8233884659CC65DF58BA2D031 DC7CC3B2BA3A97C61582CFB8FD6F4421 O 68288274 255 1\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201404\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "201405\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 910EB9954F01FCFA0A630A60CE5B5D1C O 73691286 28 1\n",
      "216E0E920C0B51093276E270CC36CB72 EE41785ADD91736F19EC2B96891004E5 I 73579125 36 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 73558700 38 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 73696911 56 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 73550777 81 5\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 73650732 87 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 73600150 78 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 73555948 83 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 73678366 204 6\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201406\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 76351355 23 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 76351325 61 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 76308688 106 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 76322196 15 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 76342742 1 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 76342773 4 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 76334335 103 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 AEDBAEB9612EE912F80B38C00AD6F9A4 O 76350993 76 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 C7F0EA3A19400F9A8D86AB541582582B I 76244998 206 2\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201407\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 93227A67F5708A0A207F3ED51EE9CA44 O 78945730 56 1\n",
      "216E0E920C0B51093276E270CC36CB72 FCD8CFEC3F5BD2EC3893ABB35A1616DF O 78959179 37 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0A107D10AF47F6BE37B6C534696EC181 O 78871953 371 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 78937448 133 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 78932770 14 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 78930608 27 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 78848859 347 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 78832364 355 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 78936149 108 2\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201408\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 7D55073541E3434592411AFC8DA61CFB I 81533407 31 1\n",
      "216E0E920C0B51093276E270CC36CB72 910EB9954F01FCFA0A630A60CE5B5D1C O 81555068 50 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0A107D10AF47F6BE37B6C534696EC181 O 81619444 153 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 0A107D10AF47F6BE37B6C534696EC181 O 81611217 180 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 81631012 387 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 81604159 0 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 81525794 0 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 81614726 134 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 81513149 18 1\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201409\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 02569B30FC6563D40BD6803416A4CCF0 O 84229336 134 1\n",
      "216E0E920C0B51093276E270CC36CB72 02569B30FC6563D40BD6803416A4CCF0 O 84231991 23 2\n",
      "216E0E920C0B51093276E270CC36CB72 02569B30FC6563D40BD6803416A4CCF0 O 84231991 23 2\n",
      "216E0E920C0B51093276E270CC36CB72 02569B30FC6563D40BD6803416A4CCF0 O 84231991 23 2\n",
      "216E0E920C0B51093276E270CC36CB72 5492B0033211A161BA4DCECD83A6690E O 84233806 104 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 0429CE90DF689F6404FDC51B13887547 O 84286920 69 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 I 84224677 462 5\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 I 84224677 462 5\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 I 84224677 462 5\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201410\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 86806984 146 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 86908423 151 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 86789375 453 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 86775857 11 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 86867214 119 5\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 86798000 166 5\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 86884496 17 6\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 86793846 38 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 86867233 3 5\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201411\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 89477937 63 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 89544313 129 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 89542574 71 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 C7F0EA3A19400F9A8D86AB541582582B O 89555968 106 2\n",
      "46465E22CBFF2834A125FE54C44563E3 32B8A1B34A851ADC4BE18A7CF7D8062B O 89462679 129 3\n",
      "46465E22CBFF2834A125FE54C44563E3 54F254B0BF26C43E92A7B3AB1CBB1591 I 89587619 126 4\n",
      "46465E22CBFF2834A125FE54C44563E3 54F254B0BF26C43E92A7B3AB1CBB1591 I 89500775 42 4\n",
      "46465E22CBFF2834A125FE54C44563E3 806986227C93BE2E3152014A91958282 O 89461853 183 4\n",
      "46465E22CBFF2834A125FE54C44563E3 806986227C93BE2E3152014A91958282 O 89560788 67 5\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201412\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 92150868 394 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 92150268 600 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 92048220 18 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 92158303 287 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 92048012 43 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 92072504 370 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 92049280 46 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 92151201 48 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 77B47CC119CE1E86ED0F00F090D59D2F I 92167174 544 3\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201501\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 5492B0033211A161BA4DCECD83A6690E O 94695418 88 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 I 94820900 153 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 94842758 189 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 94820507 361 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 94831395 52 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 29D24A2CD146FFEDC05757A22BAED052 I 94835642 40 5\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 94711785 25 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 94838044 27 6\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 94819123 34 6\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201502\n",
      "\n",
      "201503\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "216E0E920C0B51093276E270CC36CB72 1A1FFCB17EEB2ACB08724C00153CFD4E I 99868716 13 1\n",
      "216E0E920C0B51093276E270CC36CB72 D3741461B9135EAB924F3E269DBE1919 O 99945593 158 2\n",
      "216E0E920C0B51093276E270CC36CB72 D3741461B9135EAB924F3E269DBE1919 O 99946422 301 1\n",
      "216E0E920C0B51093276E270CC36CB72 D3741461B9135EAB924F3E269DBE1919 O 99944993 600 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 99924275 133 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 99920151 372 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 99932979 7 4\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 99933107 282 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 99923805 66 3\n",
      "\n",
      "gzip: stdout: Broken pipe\n",
      "201504\n",
      "USER OTHER_USER DIRECTION TIMESTAMP DURATION ANTENNA_ID\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 102528667 46 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 0ECA1BB365D1BD1F3B54F5F5225A4337 O 102513950 98 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 102613223 105 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 102522566 192 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 102588263 68 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 102596243 73 2\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 I 102611736 77 1\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 102605786 140 3\n",
      "73C2C1F8233884659CC65DF58BA2D031 3386587CF389FB2B195C60D92DEB9F81 O 102605786 140 4\n",
      "\n",
      "gzip: stdout: Broken pipe\n"
     ]
    }
   ],
   "source": [
    "for month in months:\n",
    "    input_file= rootdir +\"simple_format_{m}.txt.gz\"\\\n",
    "            .format(m=month)\n",
    "    print(month)\n",
    "    !zcat $input_file | head"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "months14= [\"2014\"+ \"0\"+str(month) if int(month )<10 else \"2014\"+str(month) for month in range(1,13)]\n",
    "months15= [\"2015\"+ \"0\"+str(month) if int(month )<10 else \"2015\"+str(month) for month in range(1,5)]\n",
    "months = months14  + months15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      " working group number 0 of 15, time elapsed is 0.00229692459106 \n",
      "\n"
     ]
    },
    {
     "ename": "CParserError",
     "evalue": "Error tokenizing data. C error: Calling read(nbytes) on source failed. Try engine='python'.",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mCParserError\u001b[0m                              Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-142-6b795b36863b>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m     44\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     45\u001b[0m         \u001b[0mnumb\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m0\u001b[0m\u001b[1;33m;\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 46\u001b[1;33m         \u001b[1;32mfor\u001b[0m \u001b[0mchunk\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtable\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     47\u001b[0m                 \u001b[0mnumb\u001b[0m\u001b[1;33m+=\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     48\u001b[0m                 \u001b[1;31m#aca filtro por la cantidad\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/home/juan/anaconda2/lib/python2.7/site-packages/pandas/io/parsers.pyc\u001b[0m in \u001b[0;36m__iter__\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    721\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mchunksize\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    722\u001b[0m                 \u001b[1;32mwhile\u001b[0m \u001b[0mTrue\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 723\u001b[1;33m                     \u001b[1;32myield\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mchunksize\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    724\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    725\u001b[0m                 \u001b[1;32myield\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/home/juan/anaconda2/lib/python2.7/site-packages/pandas/io/parsers.pyc\u001b[0m in \u001b[0;36mread\u001b[1;34m(self, nrows)\u001b[0m\n\u001b[0;32m    745\u001b[0m                 \u001b[1;32mraise\u001b[0m \u001b[0mValueError\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'skip_footer not supported for iteration'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    746\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 747\u001b[1;33m         \u001b[0mret\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_engine\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mnrows\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    748\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    749\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0moptions\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mget\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'as_recarray'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/home/juan/anaconda2/lib/python2.7/site-packages/pandas/io/parsers.pyc\u001b[0m in \u001b[0;36mread\u001b[1;34m(self, nrows)\u001b[0m\n\u001b[0;32m   1195\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mread\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mnrows\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0mNone\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1196\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m-> 1197\u001b[1;33m             \u001b[0mdata\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_reader\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mread\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mnrows\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m   1198\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mStopIteration\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1199\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_first_chunk\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mpandas/parser.pyx\u001b[0m in \u001b[0;36mpandas.parser.TextReader.read (pandas/parser.c:7988)\u001b[1;34m()\u001b[0m\n",
      "\u001b[1;32mpandas/parser.pyx\u001b[0m in \u001b[0;36mpandas.parser.TextReader._read_low_memory (pandas/parser.c:8444)\u001b[1;34m()\u001b[0m\n",
      "\u001b[1;32mpandas/parser.pyx\u001b[0m in \u001b[0;36mpandas.parser.TextReader._read_rows (pandas/parser.c:8970)\u001b[1;34m()\u001b[0m\n",
      "\u001b[1;32mpandas/parser.pyx\u001b[0m in \u001b[0;36mpandas.parser.TextReader._tokenize_rows (pandas/parser.c:8838)\u001b[1;34m()\u001b[0m\n",
      "\u001b[1;32mpandas/parser.pyx\u001b[0m in \u001b[0;36mpandas.parser.raise_parser_error (pandas/parser.c:22649)\u001b[1;34m()\u001b[0m\n",
      "\u001b[1;31mCParserError\u001b[0m: Error tokenizing data. C error: Calling read(nbytes) on source failed. Try engine='python'."
     ]
    }
   ],
   "source": [
    "\n",
    "## script para extraer los labels de los simple_formats de 2014\n",
    "#cada dataset tiene aprox 450M de datos, de los cuales aprox 45M son de los Telco users que nos intersan\n",
    "# como hay 16 meses a analizar ---> tengo 720M registros que mirar, luego hago 15 pasadas filtrados por el user_hash\n",
    "\n",
    "#ver el tiempo que tarda\n",
    "start_time = time.time()\n",
    "\n",
    "#el chunk basicamente va leyendo el file de a 'chunksize' cantidad de filas\n",
    "#subgroup = pd.DataFrame()\n",
    "\n",
    "#hay 12 meses del cual extraer atributos:\n",
    "months14= [\"2014\"+ \"0\"+str(month) if int(month )<10 else \"2014\"+str(month) for month in range(1,13)]\n",
    "months15= [\"2015\"+ \"0\"+str(month) if int(month )<10 else \"2015\"+str(month) for month in range(1,5)]\n",
    "months = months14  + months15\n",
    "\n",
    "#aca seteamos como vamos a partir la tabla segun el nro correspondiente a c/hash y tomando modulo\n",
    "passes = 15\n",
    "#creo los grupos que despues van a filtrar c/chunk de la tabla para hacer varias pasadas\n",
    "for group in range(passes):\n",
    "    print(\"\\n working group number {it} of {pas}, time elapsed is {t} \\n\".\\\n",
    "          format(it=group,pas=passes, t=(time.time()-start_time)))\n",
    "\n",
    "    #itero sobre los meses\n",
    "    subgroup = pd.DataFrame()\n",
    "    for month in months:\n",
    "        if month == '201404':\n",
    "            continue\n",
    "            \n",
    "        input_file= rootdir +\"simple_format_{m}.txt.gz\"\\\n",
    "            .format(y=year,m=month)\n",
    "\n",
    "        table = pd.read_csv(\n",
    "                input_file,\n",
    "                engine = 'c',\n",
    "                chunksize = 11*10**7,\n",
    "    #            iterator =True,\n",
    "                sep = ' ',\n",
    "                header = 0,\n",
    "                index_col=None,\n",
    "                usecols = ['USER','ANTENNA_ID'],\n",
    "                error_bad_lines= False,\n",
    "                warn_bad_lines= True,\n",
    "                dtype = {'ANTENNA_ID':np.uint16,'USER':np.object_}\n",
    "                )\n",
    "\n",
    "        #cuando entramos a este loop, table tiene tantos 'chunks' como el valor entero de la cantidad de lineas en el file\n",
    "        #dividido el tamanyo del chunksize\n",
    "\n",
    "        numb=0;\n",
    "        for chunk in table:\n",
    "                numb+=1\n",
    "                #aca filtro por la cantidad \n",
    "                chunk = chunk[chunk['USER'].isin(user_hashes.loc[:,'LineKeyOrigin'].values)]\n",
    "                #a cada chunk filtro por todos los USERs por el hash del string (el hash es un int) y despue filtro modulo passes\n",
    "                #y trabajo sobre la tabla subgroup nada mas que ahora tiene menos usuarios\n",
    "                subgroup = subgroup.append(chunk[chunk['USER'].apply(lambda x: hash(x) % passes == group )])\n",
    "    \n",
    "        print(\"Finished parsing month {ms}-{y}, time elapsed is {t} \".format(ms=month,y=year,t=(time.time()-start_time)))\n",
    "        \n",
    "    #entonces la idea es que yo ahora solo voy a trabajar, dentro de esta tabla filtrada y para todos los meses juntos\n",
    "    \n",
    "    #paso de segundos a horas\n",
    "    #notar que TIMESTAMP arranca en 0 segundos para domingo 01/01/2012 00:00am \n",
    "    #con lo cual domingo es el dia 0, lunes el 1, asi..\n",
    "    print('finished reading months for group {it} of {pas}: time elapsed is {t} \\n'.\\\n",
    "    format(it=group,pas=passes, t=(time.time()-start_time)) )\n",
    "    \n",
    "    #filtro usuarios con pocos o demasiados llamados en general menos de 25 y mas de 2000 en 5 meses  \n",
    "    grouped = subgroup.groupby(['USER', 'ANTENNA_ID'])['ANTENNA_ID'].agg({'count': np.size})\n",
    "    \n",
    "    del subgroup\n",
    "    grouped.reset_index(inplace=True,drop=False)   \n",
    "   \n",
    "    #reordeno dentro de c/ USER por el count del antenna, esto me sirve para despues ordenar las antenas por uso\n",
    "    grouped.sort_values(by=['USER','count'],ascending=False,inplace=True)\n",
    "    grouped = pd.merge(grouped,antennas['EPIDEMIC'].reset_index(), on='ANTENNA_ID',how = 'left')\n",
    "\n",
    "    ##enriquezco la muestra con datos epidemicos\n",
    "    #primero agrego a cada antenna del df el dato de si es epidemica\n",
    "    #despues agrupo por el USER y me fijo solo la columna epidemica en c/grupo\n",
    "    #finalmente sumo en c/ grupo y tomo la parte superior \n",
    "    #entera de esa division con el largo del grupo. Si uso al menos una antena epidemica entonces esta expuesto(==1) Si no,\n",
    "    # da 0 pues no estuvo expuesto.        \n",
    "    ##enriquezco la muestra con datos epidemicos\n",
    "\n",
    "    exposed_info = grouped.groupby('USER')['EPIDEMIC'].\\\n",
    "        agg({'EXPOSED' : lambda x: int( np.ceil(np.sum(x)*1.0/np.size(x)) )})\n",
    "\n",
    "    #actualizo la tabla grouped\n",
    "    grouped = grouped.join(exposed_info['EXPOSED'],on=\"USER\",how=\"left\")\n",
    "    del exposed_info\n",
    "\n",
    "    #creo la tabla filtrada solo por users, que es la que voy a terminar guardando (hay tantos rows como users)\n",
    "    output_table = grouped.drop_duplicates(subset = 'USER', keep='first')\n",
    "    del grouped\n",
    "\n",
    "    #re indexo\n",
    "    output_table.set_index('USER',inplace=True,drop=False)\n",
    "    \n",
    "    #agrupo ahora la tabla por USER para hacer todos los calculos en los grupos\n",
    "    output_table.drop('count',axis=1,inplace=True)\n",
    "    output_table = output_table.set_index('USER')\n",
    "    #paso a ints todas las columnas (el hash ya esta de index)\n",
    "    output_table = output_table.astype(int,copy=False)\n",
    "    \n",
    "    #aca termino guardando (en forma de append) el output final pero solo para esos usuarios % pass ==group\n",
    "\n",
    "    #el primer write (first group==0) va con header\n",
    "    if group ==0 :\n",
    "        output_table.to_csv(output_file, index = True,\n",
    "                   header = True,sep = \"|\", \n",
    "                            compression = \"gzip\", mode='w')\n",
    "    else: \n",
    "        output_table.to_csv(output_file, index = True,\n",
    "                   header = False,sep = \"|\", \n",
    "                    compression = \"gzip\", mode='a')\n",
    "\n",
    "    print(\"Finished group {gr} of {ps}, time elapsed is {t}\\n \".format(gr=group,ps=passes,t=(time.time()-start_time)))\n",
    "\n",
    "print(\"total running time of script is %d \" % time.time() - start_time)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## cambiamos el working dir y el input_file (output.txt.gz)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}