<commit_before><commit_msg>added module to generate presumOverallRatings for all seasons<commit_after>{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark import SparkContext \n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.types import *         # for defining schema with various datatypes\n",
    "import pyspark.sql.functions as func    # for ETL, data processing on Dataframes\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "sc = SparkContext()        # creating sparkcontext\n",
    "sql = SQLContext(sc)       # creating SQLcontext\n",
    "\n",
    "data_opath = \"../output/csv/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getDF(ratingType, filename):\n",
    "    resultRDD = sc.textFile(data_opath + filename)\n",
    "    resultHeader = resultRDD.filter(lambda l : ratingType in l)\n",
    "    resultRDDNoHeader = resultRDD.subtract(resultHeader)\n",
    "    if(ratingType == \"batsman\"):\n",
    "        resultTempRDD = resultRDDNoHeader.map(lambda k : k.split(',')).map(lambda p: (p[0],p[1], int(p[2]), int(p[3]),int(p[4]), int(p[5]), int(p[6]),int(p[7]), int(p[8]), round(float(p[9]),2)))\n",
    "    elif(ratingType == \"bowler\"):\n",
    "        resultTempRDD = resultRDDNoHeader.map(lambda k : k.split(',')).map(lambda p: (p[0],p[1], int(p[2]), int(p[3]),int(p[4]), int(p[5]), int(p[6]),int(p[7]), int(p[8]), int(p[9]), int(p[10]), round(float(p[11]),2)))\n",
    "    elif(ratingType == \"fielder\"):\n",
    "        resultTempRDD = resultRDDNoHeader.map(lambda k : k.split(',')).map(lambda p: (p[0],p[1], int(p[2]), int(p[3])))\n",
    "   \n",
    "    resultDF = sql.createDataFrame(resultTempRDD, resultRDD.first().split(','))\n",
    "    resultDF = resultDF.drop('')\n",
    "    resultDF = resultDF.dropDuplicates([ratingType])\n",
    "    return resultDF\n",
    "\n",
    "\n",
    "def getInitDF(ratingType, seasonLBound, seasonUBound):\n",
    "    filename = ratingType+str(seasonLBound)+\"_\"+str(seasonUBound)+\".csv\"\n",
    "    resultDF = getDF(ratingType, filename)\n",
    "    resultDF = resultDF.select(ratingType)\n",
    "    return resultDF.sort(ratingType)\n",
    "\n",
    "\n",
    "def getSeasonRangeRatingsDF(ratingType,seasonNum):\n",
    "    filename = ratingType+str(seasonNum)+\".csv\"\n",
    "    resultDF = getDF(ratingType, filename)\n",
    "    resultDF = resultDF.select(ratingType, \"overall\")\n",
    "    return resultDF.sort(ratingType)\n",
    "\n",
    "\n",
    "def fixNoneValues(score):\n",
    "    if(score == None):\n",
    "        return 0\n",
    "    else:\n",
    "        return score\n",
    "\n",
    "\n",
    "def getPreSum(currOverall, prevPreSum):\n",
    "    return currOverall+prevPreSum\n",
    "\n",
    "\n",
    "def joinOverallDF(aDF, bDF, colName):\n",
    "    return aDF.join(bDF, ratingType, \"left\").withColumnRenamed(\"overall\", colName)\n",
    "\n",
    "\n",
    "def fixJoinOverallDF(srcDF, colName):\n",
    "    fixNoneValuesUDF = udf(fixNoneValues, LongType())\n",
    "    return srcDF.withColumn(colName,fixNoneValuesUDF(srcDF[colName]))\n",
    "\n",
    "\n",
    "def calcPreSumOverall(srcDF, seasonLBound, seasonNum, ratingType):\n",
    "    getPreSumUDF = udf(getPreSum, LongType())\n",
    "    idx = seasonNum - seasonLBound + 1\n",
    "    if(1 == idx):\n",
    "        return srcDF\n",
    "    else:\n",
    "        currColName = ratingType+\"presumoverall\"+str(seasonNum)\n",
    "        prevColName = ratingType+\"presumoverall\"+str(seasonNum-1)\n",
    "        srcDF = srcDF.withColumn(currColName, getPreSumUDF(srcDF[currColName], srcDF[prevColName]))\n",
    "        return srcDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratingType = \"fielder\"\n",
    "seasonLBound = 2008\n",
    "seasonUBound = 2016\n",
    "finalDF = getInitDF(ratingType, seasonLBound, seasonUBound)\n",
    "\n",
    "for i in range(seasonLBound, seasonUBound+1):\n",
    "    srcDF = getSeasonRangeRatingsDF(ratingType, i)\n",
    "    colName = ratingType+\"presumoverall\"+str(i)\n",
    "    finalDF = joinOverallDF(finalDF, srcDF, colName)\n",
    "    finalDF = fixJoinOverallDF(finalDF, colName)\n",
    "    finalDF = calcPreSumOverall(finalDF, seasonLBound, i, ratingType)\n",
    "\n",
    "finalDF = finalDF.sort(ratingType+\"presumoverall\"+str(seasonUBound), ascending=0)\n",
    "filename = ratingType+\"preSumOverall\"+str(seasonLBound)+\"_\"+str(seasonUBound)+\".csv\"\n",
    "finalDF.toPandas().to_csv(data_opath + filename)"
   ]
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
{'fraction_non_alphanumeric': 0.20486241990199774, 'fraction_numerical': 0.011119487372785525, 'mean_word_length': 3.0729086722947043, 'pattern_counts': {'":': 34, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3681630', 'n_tokens_mistral': 2033, 'n_tokens_neox': 1981, 'n_words': 320}