csvParser.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import pandas\n",
    "from sqlalchemy import create_engine\n",
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "class csvParser:\n",
    "    \n",
    "    data = pandas.DataFrame()\n",
    "    config = {}\n",
    "    type_map = {\n",
    "            'float':float,\n",
    "            'int':int,\n",
    "            'str':str\n",
    "    }\n",
    "    \n",
    "    def __init__(self, confPath, auto=True):\n",
    "        self.op_map = {\n",
    "            \"remove\":{\"func\":self.removeCols,\"args\":1},\n",
    "            \"hash\":{\"func\":self.hashCols,\"args\":1},\n",
    "            \"cast\":{\"func\":self.castCols,\"args\":1},\n",
    "            \"sample\":{\"func\":self.sampleRows, \"args\":1}\n",
    "        }\n",
    "        \n",
    "        self.loadConfig(confPath)\n",
    "        if (auto==True):\n",
    "            self.loadData()\n",
    "            self.processData()\n",
    "            self.storeData()\n",
    "    \n",
    "    def loadConfig(self,path):\n",
    "        try:\n",
    "            with open(path, 'r') as config:\n",
    "                data=config.read().replace('\\n', '')\n",
    "            self.config = json.loads(data)\n",
    "            return True\n",
    "        except:\n",
    "            return False\n",
    "        \n",
    "    def loadData(self):\n",
    "        try:\n",
    "            filepath = self.config['filepath']\n",
    "            print(\"loading data from \"+filepath)\n",
    "            col_dtypes = self.config['load_cols']\n",
    "            cols = list(col_dtypes.keys())  \n",
    "            self.data = pandas.read_csv(filepath, usecols = cols, dtype = col_dtypes)  \n",
    "            self.data.columns = list(map(lambda x: x.strip(), self.data.columns))\n",
    "            return True\n",
    "        except:\n",
    "            return False\n",
    "\n",
    "    def processData(self):\n",
    "        for step in self.config['processing_steps']:\n",
    "            op = list(step.keys())[0]\n",
    "            params = list(step.values())[0]\n",
    "            self.processDataStep(op,params)\n",
    "            \n",
    "    def processDataStep(self,op,params):\n",
    "        if (type(params)!=list):\n",
    "            params = [params]\n",
    "        f = self.op_map[op]['func']\n",
    "        arg_num = self.op_map[op]['args']\n",
    "        if (arg_num==1):\n",
    "            f(params)\n",
    "        elif (arg_num==2):\n",
    "            ps = params.items()\n",
    "            f(ps[0], ps[1])\n",
    "        \n",
    "    def removeCols(self, names):\n",
    "        for name in names:\n",
    "            try:\n",
    "                print(\"removing col=\"+name)\n",
    "                del self.data[name]\n",
    "            except:\n",
    "                print(\"failed to del\" + name)\n",
    "                return False\n",
    "        return True\n",
    "    \n",
    "    def castCols(self, type_castings):\n",
    "        for cast in type_castings:\n",
    "            for col in cast.keys():\n",
    "                print(\"casting col=\"+col+\" to type=\"+cast[col])\n",
    "                self.data[col] = self.data[col].astype(self.type_map[cast[col]])\n",
    "        return True\n",
    "    \n",
    "    def hashCols(self, names):\n",
    "        #converts col contents to string before hashing\n",
    "        for name in names:\n",
    "            print(\"hashing col=\"+name)\n",
    "            self.data[name] = self.data[name].apply(lambda x: hash(str(x)))\n",
    "        return True\n",
    "    \n",
    "    def sampleRows(self,perc):\n",
    "        print(\"sampling \"+str(perc[0]*100)+\"% of data\")\n",
    "        self.data = self.data.sample(frac=perc[0])\n",
    "        return True\n",
    "    \n",
    "    def storeData(self):\n",
    "        dbname = self.config['sql']['db']\n",
    "        tablename = self.config['sql']['table']\n",
    "        print(\"storing data in table=\"+tablename+\" in db=\"+dbname)\n",
    "        cols = list(self.data.columns)\n",
    "        engine = create_engine(dbname)\n",
    "        self.data.to_sql(tablename, engine)\n",
    "        return True\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "testParser = csvParser(\"./csvParse.config\", auto=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'filepath': './testData.csv',\n",
       " 'load_cols': {'col1': 'float', 'col2': 'int', 'col3': 'int', 'col4': 'str'},\n",
       " 'processing_steps': [{'remove': ['col1']},\n",
       "  {'hash': ['col2', 'col4']},\n",
       "  {'cast': {'col3': 'float'}},\n",
       "  {'sample': 0.5}],\n",
       " 'sql': {'db': 'sqlite:////data/quartet/testParse.db', 'table': 'testCsv'}}"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testParser.config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loading data from ./testData.csv\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>col1</th>\n",
       "      <th>col2</th>\n",
       "      <th>col3</th>\n",
       "      <th>col4</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>2</td>\n",
       "      <td>3</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>4</td>\n",
       "      <td>5</td>\n",
       "      <td>6</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   col1  col2  col3 col4\n",
       "0     1     2     3    a\n",
       "1     4     5     6    b"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testParser.loadData()\n",
    "testParser.data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "removing col=col1\n",
      "hashing col=col2\n",
      "hashing col=col4\n",
      "casting col=col3 to type=float\n",
      "sampling 50.0% of data\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>col2</th>\n",
       "      <th>col3</th>\n",
       "      <th>col4</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>-2480217190145936623</td>\n",
       "      <td>6</td>\n",
       "      <td>6106447415896647459</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                  col2  col3                 col4\n",
       "1 -2480217190145936623     6  6106447415896647459"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testParser.processData()\n",
    "testParser.data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dtype('float64')"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testParser.data['col3'].dtype"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "storing data in table=testCsv in db=sqlite:////data/quartet/testParse.db\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testParser.storeData()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#FUTURE:\n",
    "\n",
    "- Currently headers aren't stripped of whitespace before being compared to config, it would be nice to fix this\n",
    "- More datatypes (ex datetime)\n",
    "- Seperate out loading, processing, and storing into seperate classes for better modularity\n",
    "- More input types (ex load from db)\n",
    "- More output types (ex csv)\n",
    "- More transform types (ex round)\n",
    "- Save transformed cols as new cols instead of overwriting (ex col3_float)\n",
    "- Better error handling"
   ]
  }
 ],
 "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.4.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}