FinalAnalysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Importing Libraries\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.master(\"local\").appName('Ops').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "path = \"E:/Rutgers/Projects/MDSR/IPL-MSDR\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Final Analysis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reading data\n",
    "matches = spark.read.csv(path + '/dataset/clean_data/matches.csv',inferSchema=True,header=True)\n",
    "deliveries = spark.read.csv(path + '/dataset/clean_data/deliveries.csv',inferSchema=True,header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating temporary tables of the data\n",
    "matches.registerTempTable('matches_db')\n",
    "deliveries.registerTempTable('deliveries_db')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Merging both the tables\n",
    "merged_db = spark.sql('select m.*,d.* from matches_db as m inner join deliveries_db as d on m.id=d.match_id')\n",
    "merged_db.registerTempTable('analysis_db')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Batting Metrics "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nmba: no. of batsmen\n",
    "# nm: no. of matches played by a batsman\n",
    "# hha: hard hitting ability\n",
    "# f: finisher\n",
    "# fsa: fast scoring ability\n",
    "# con: consistency\n",
    "# rbw: running between wickets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|No_of_Batsman|\n",
      "+-------------+\n",
      "|          516|\n",
      "+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Number of Batsmen\n",
    "nmba = spark.sql('select count(distinct(batsman)) as No_of_Batsman from analysis_db')\n",
    "nmba.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of Matches played by a batsmen\n",
    "nm = spark.sql('select batsman, count(distinct(match_id)) as No_of_Matches \\\n",
    "                from analysis_db group by batsman')\n",
    "nm.registerTempTable('no_of_matches_table')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Hard Hitting Ability "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Hard Hitting Ability = (4*Fours + 6*Sixes)/Balls Played by Batsman\n",
    "hha = spark.sql('select nmt.batsman as Batsman, round(nvl(t4.hard_hitting_ability,0), 5) as \\\n",
    "                Hard_Hitting_Ability from \\\n",
    "                (select t1.batsman, (t1.fours*4 + t2.sixes*6)/t3.balls_played as hard_hitting_ability\\\n",
    "                from (select batsman,count(*) as fours from analysis_db where batsman_runs = 4 group by batsman) t1 \\\n",
    "                inner join  \\\n",
    "                (select batsman,count(*) as sixes from analysis_db where batsman_runs = 6 \\\n",
    "                group by batsman) t2 on t1.batsman=t2.batsman\\\n",
    "                inner join\\\n",
    "                (select batsman,count(*) as balls_played from analysis_db \\\n",
    "                group by batsman) t3 on t3.batsman=t1.batsman) t4 \\\n",
    "                right join \\\n",
    "                no_of_matches_table nmt on t4.batsman = nmt.batsman')\n",
    "hha.registerTempTable('hard_hitting_ability')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "hha = spark.sql('select rank() over (order by Hard_Hitting_Ability desc) as Rank, t1.* \\\n",
    "                from hard_hitting_ability t1 \\\n",
    "                inner join \\\n",
    "                no_of_matches_table t2\\\n",
    "                on t1.batsman = t2.batsman where no_of_matches>9')\n",
    "hha.registerTempTable('hard_hitting_ability')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------+--------------------+-------+\n",
      "|Rank|      Batsman|Hard_Hitting_Ability|Weights|\n",
      "+----+-------------+--------------------+-------+\n",
      "|   1|   AD Russell|             1.37733|0.99583|\n",
      "|   2|    SP Narine|             1.33056|0.99167|\n",
      "|   3|        M Ali|             1.21311| 0.9875|\n",
      "|   4|    KK Cooper|                 1.2|0.98333|\n",
      "|   5|  BCJ Cutting|             1.19178|0.97917|\n",
      "|   6|    K Gowtham|             1.16279|  0.975|\n",
      "|   7|CR Brathwaite|             1.13333|0.97083|\n",
      "|   8|     CH Gayle|             1.10699|0.96667|\n",
      "|   9|  Rashid Khan|             1.10448| 0.9625|\n",
      "|  10|   GJ Maxwell|             1.09313|0.95833|\n",
      "+----+-------------+--------------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "hha = spark.sql('select t1.*, round((240-rank)/240, 5) as Weights \\\n",
    "                from hard_hitting_ability t1')\n",
    "hha.registerTempTable('hard_hitting_ability')\n",
    "hha.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Finisher "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Finisher = Not Out innings/Total Innings played\n",
    "f = spark.sql('select t3.batsman as Batsman, round(t3.not_out_innings/t4.total_matches_played, 5) as Finisher from\\\n",
    "              (select t1.batsman, t1.matches_played-t2.number_of_times_out as not_out_innings from \\\n",
    "              (select batsman, count(distinct(match_id)) as matches_played from analysis_db group by batsman) t1\\\n",
    "              inner join \\\n",
    "              (select batsman, count(*) as number_of_times_out from analysis_db where player_dismissed = batsman \\\n",
    "              group by batsman) t2\\\n",
    "              on t1.batsman=t2.batsman) t3\\\n",
    "              inner join\\\n",
    "              (select batsman, count(distinct(match_id)) as total_matches_played \\\n",
    "              from analysis_db group by batsman) t4\\\n",
    "              on t3.batsman = t4.batsman')\n",
    "f.registerTempTable('finisher')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "f = spark.sql('select rank() over (order by finisher desc) as Rank, t1.* \\\n",
    "              from finisher t1 \\\n",
    "              inner join \\\n",
    "              no_of_matches_table t2 \\\n",
    "              on t1.batsman = t2.batsman \\\n",
    "              where no_of_matches>9')\n",
    "f.registerTempTable('finisher')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------+--------+-------+\n",
      "|Rank|       Batsman|Finisher|Weights|\n",
      "+----+--------------+--------+-------+\n",
      "|   1| Iqbal Abdulla| 0.92308|0.99583|\n",
      "|   2|      A Kumble| 0.86667|0.99167|\n",
      "|   3|Sandeep Sharma| 0.78571| 0.9875|\n",
      "|   4|   S Sreesanth|    0.75|0.98333|\n",
      "|   5|     S Aravind|     0.7|0.97917|\n",
      "|   5|     JJ Bumrah|     0.7|0.97917|\n",
      "|   5|      VR Aaron|     0.7|0.97917|\n",
      "|   8|     YS Chahal| 0.66667|0.96667|\n",
      "|   8|      I Sharma| 0.66667|0.96667|\n",
      "|  10|  Bipul Sharma| 0.64706|0.95833|\n",
      "+----+--------------+--------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "f = spark.sql('select t1.*, round((240-rank)/240, 5) as Weights \\\n",
    "              from finisher t1')\n",
    "f.registerTempTable('finisher')\n",
    "f.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fast Scoring Ability"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fast Scoring Ability = Total Runs/Balls Played by Batsman\n",
    "fsa = spark.sql('select batsman as Batsman, round(Total_Runs/balls_played, 5) as Fast_Scoring_Ability \\\n",
    "                  from (select batsman,sum(batsman_runs) as Total_Runs, count(*) as balls_played \\\n",
    "                  from analysis_db group by batsman)')\n",
    "fsa.registerTempTable('fast_scoring_ability')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "fsa = spark.sql('select rank() over (order by fast_scoring_ability desc) as Rank, t1.* \\\n",
    "                  from fast_scoring_ability t1 \\\n",
    "                  inner join \\\n",
    "                  no_of_matches_table t2 \\\n",
    "                  on t1.batsman = t2.batsman where no_of_matches>9')\n",
    "fsa.registerTempTable('fast_scoring_ability')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------+--------------------+-------+\n",
      "|Rank|      Batsman|Fast_Scoring_Ability|Weights|\n",
      "+----+-------------+--------------------+-------+\n",
      "|   1|   AD Russell|              1.7995|0.99583|\n",
      "|   2|    K Gowtham|             1.72093|0.99167|\n",
      "|   3|        M Ali|             1.69945| 0.9875|\n",
      "|   4|    SP Narine|             1.66944|0.98333|\n",
      "|   5|    KK Cooper|             1.65714|0.97917|\n",
      "|   6|  BCJ Cutting|             1.64384|  0.975|\n",
      "|   7|  Rashid Khan|             1.62687|0.97083|\n",
      "|   8|      RR Pant|             1.62319|0.96667|\n",
      "|   9|   J Bairstow|             1.59727| 0.9625|\n",
      "|  10|CR Brathwaite|             1.56667|0.95833|\n",
      "+----+-------------+--------------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fsa = spark.sql('select t1.*, round((240-rank)/240, 5) as Weights \\\n",
    "                from fast_scoring_ability t1')\n",
    "fsa.registerTempTable('fast_scoring_ability')\n",
    "fsa.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Consistency"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Consistency = Total Runs/Number of Times Out\n",
    "con = spark.sql('select t1.batsman as Batsman, round(t1.Total_runs/t2.no_of_times_dismissed, 5) as Consistency \\\n",
    "                from (select batsman,sum(batsman_runs) as Total_runs \\\n",
    "                from analysis_db group by batsman) t1 \\\n",
    "                inner join \\\n",
    "                (select batsman, count(*) as no_of_times_dismissed \\\n",
    "                from analysis_db where player_dismissed is not null \\\n",
    "                group by batsman) t2 on t1.batsman=t2.batsman')\n",
    "con.registerTempTable('consistency')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "con = spark.sql('select rank() over (order by consistency desc) as Rank, t1.* \\\n",
    "                  from consistency t1 \\\n",
    "                  inner join \\\n",
    "                  no_of_matches_table t2 \\\n",
    "                  on t1.batsman = t2.batsman where no_of_matches>9')\n",
    "con.registerTempTable('consistency')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------+-----------+-------+\n",
      "|Rank|      Batsman|Consistency|Weights|\n",
      "+----+-------------+-----------+-------+\n",
      "|   1|   AD Russell|     1.7995|0.99583|\n",
      "|   2|    K Gowtham|    1.72093|0.99167|\n",
      "|   3|        M Ali|    1.69945| 0.9875|\n",
      "|   4|    SP Narine|    1.66944|0.98333|\n",
      "|   5|    KK Cooper|    1.65714|0.97917|\n",
      "|   6|  BCJ Cutting|    1.64384|  0.975|\n",
      "|   7|  Rashid Khan|    1.62687|0.97083|\n",
      "|   8|      RR Pant|    1.62319|0.96667|\n",
      "|   9|   J Bairstow|    1.59727| 0.9625|\n",
      "|  10|CR Brathwaite|    1.56667|0.95833|\n",
      "+----+-------------+-----------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "con = spark.sql('select t1.*, round((240-Rank)/240, 5) as Weights \\\n",
    "                from consistency t1')\n",
    "con.registerTempTable('consistency')\n",
    "con.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Running Between Wickets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Running Between Wickets = (Total Runs – (4*Fours + 6*Sixes))/(Total Balls Played – Boundary Balls)\n",
    "rbw = spark.sql('select t9.batsman as Batsman, round(nvl(t8.running_between_wickets,0), 5) as Running_Between_Wickets \\\n",
    "                from (select t4.batsman, t4.first_bracket/t7.second_bracket as Running_Between_Wickets \\\n",
    "                from (select t1.batsman, t3.total_runs-(t1.fours*4 + t2.sixes*6) as first_bracket \\\n",
    "                from (select batsman,count(*) as fours from analysis_db where batsman_runs = 4 \\\n",
    "                group by batsman) t1 \\\n",
    "                inner join \\\n",
    "                (select batsman,count(*) as sixes from analysis_db where batsman_runs = 6 group by batsman) t2 \\\n",
    "                on t1.batsman=t2.batsman \\\n",
    "                inner join \\\n",
    "                (select batsman,sum(batsman_runs) as total_runs from analysis_db group by batsman) t3 \\\n",
    "                on t3.batsman=t1.batsman) t4 \\\n",
    "                inner join\\\n",
    "                (select t5.batsman, t5.total_balls_played-t6.boundry_balls as second_bracket from \\\n",
    "                (select batsman, count(*) as total_balls_played from analysis_db group by batsman) t5 \\\n",
    "                inner join \\\n",
    "                (select batsman, count(*) as boundry_balls from analysis_db where batsman_runs=4 or batsman_runs=6 \\\n",
    "                group by batsman) t6\\\n",
    "                on t5.batsman=t6.batsman) t7 \\\n",
    "                on t4.batsman=t7.batsman) t8 \\\n",
    "                right join \\\n",
    "                no_of_matches_table t9 \\\n",
    "                on t8.batsman = t9.batsman')\n",
    "rbw.registerTempTable('running_between_wickets')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "rbw = spark.sql('select rank() over (order by running_between_wickets desc) as Rank, t1.* \\\n",
    "                  from running_between_wickets t1 \\\n",
    "                  inner join \\\n",
    "                  no_of_matches_table t2\\\n",
    "                  on t1.batsman = t2.batsman where no_of_matches>9')\n",
    "rbw.registerTempTable('running_between_wickets')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------+-----------------------+-------+\n",
      "|Rank|       Batsman|Running_Between_Wickets|Weights|\n",
      "+----+--------------+-----------------------+-------+\n",
      "|   1|  Bipul Sharma|                0.85577|0.99583|\n",
      "|   2|       TM Head|                0.83206|0.99167|\n",
      "|   3|    WPUJC Vaas|                0.80882| 0.9875|\n",
      "|   4|     V Shankar|                0.79245|0.98333|\n",
      "|   5|      M Kartik|                0.78218|0.97917|\n",
      "|   6| Mohammad Nabi|                0.77922|  0.975|\n",
      "|   7|     BA Stokes|                0.76471|0.97083|\n",
      "|   8|A Ashish Reddy|                0.76364|0.96667|\n",
      "|   8|     CH Morris|                0.76364|0.96667|\n",
      "|  10|        S Gill|                0.76235|0.95833|\n",
      "+----+--------------+-----------------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rbw = spark.sql('select t1.*, round((240-rank)/240, 5) as Weights \\\n",
    "                from running_between_wickets t1')\n",
    "rbw.registerTempTable('running_between_wickets')\n",
    "rbw.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Table Name for each Metric\n",
    "# Hard Hitting Ability: hard_hitting_ability\n",
    "# Finisher: finisher\n",
    "# Fast Scoring Ability: fast_scoring_ability\n",
    "# Consistency: consistency\n",
    "# Running Between Wickets: running_between_wickets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  Total Batting Weights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+---------------------+\n",
      "|          Batsman|Total_Batting_Weights|\n",
      "+-----------------+---------------------+\n",
      "|        CH Morris|              4.60416|\n",
      "|     Bipul Sharma|              4.55833|\n",
      "|        K Gowtham|              4.40834|\n",
      "|Washington Sundar|              4.38332|\n",
      "|        HH Pandya|              4.35417|\n",
      "|         HV Patel|              4.24999|\n",
      "|     Ankit Sharma|              4.19582|\n",
      "|      Rashid Khan|              4.14583|\n",
      "|      BCJ Cutting|              4.14167|\n",
      "|    Mohammad Nabi|              4.12916|\n",
      "|   A Ashish Reddy|                4.125|\n",
      "|          SN Khan|              4.12084|\n",
      "|   AB de Villiers|              4.11667|\n",
      "|            M Ali|              4.10417|\n",
      "|       J Bairstow|                  4.1|\n",
      "|        KH Pandya|              4.06667|\n",
      "|         M Morkel|              4.04165|\n",
      "|      MF Maharoof|              4.02916|\n",
      "|        JA Morkel|              3.99583|\n",
      "|          RR Pant|              3.97084|\n",
      "|         MS Dhoni|               3.9375|\n",
      "|         J Archer|              3.93334|\n",
      "|        KK Cooper|              3.88334|\n",
      "|           S Gill|                3.875|\n",
      "|  C de Grandhomme|              3.86249|\n",
      "| RN ten Doeschate|              3.84582|\n",
      "|        DA Miller|                  3.8|\n",
      "|          TM Head|              3.79584|\n",
      "|       AD Russell|              3.79582|\n",
      "|        YK Pathan|              3.76667|\n",
      "|       JC Buttler|              3.76665|\n",
      "|       KA Pollard|              3.71667|\n",
      "|           AJ Tye|              3.69167|\n",
      "|        V Shankar|              3.68333|\n",
      "|          B Akhil|              3.64999|\n",
      "|        R Tewatia|              3.63334|\n",
      "|      JP Faulkner|              3.62917|\n",
      "|         KL Rahul|              3.60835|\n",
      "|       MP Stoinis|              3.54166|\n",
      "|        SP Narine|              3.53749|\n",
      "|       TG Southee|              3.52085|\n",
      "|        DA Warner|              3.51667|\n",
      "|       GJ Maxwell|              3.50833|\n",
      "|    KS Williamson|              3.50833|\n",
      "|          HM Amla|              3.46667|\n",
      "|      RA Tripathi|              3.44584|\n",
      "|  Harbhajan Singh|              3.37084|\n",
      "|         SK Raina|              3.34584|\n",
      "|     KP Pietersen|              3.31667|\n",
      "|           P Shaw|              3.27917|\n",
      "|          OA Shah|              3.27499|\n",
      "|    CR Brathwaite|              3.25832|\n",
      "|         SA Yadav|              3.23751|\n",
      "|          MS Gony|               3.2125|\n",
      "|        STR Binny|              3.20833|\n",
      "|      NLTC Perera|              3.20001|\n",
      "|         V Sehwag|              3.18334|\n",
      "|      Sachin Baby|              3.17916|\n",
      "|          WP Saha|              3.17083|\n",
      "|     MC Henriques|              3.16251|\n",
      "|         DJ Bravo|              3.15416|\n",
      "|       KD Karthik|              3.12917|\n",
      "|        SR Watson|              3.11251|\n",
      "|        BA Stokes|              3.11251|\n",
      "|         JR Hopes|               3.1125|\n",
      "|          CA Lynn|              3.10832|\n",
      "|       MV Boucher|              3.09583|\n",
      "|        SPD Smith|              3.09583|\n",
      "|         CH Gayle|              3.07918|\n",
      "|        KM Jadhav|              3.06665|\n",
      "|          MJ Lumb|              3.05834|\n",
      "|        SV Samson|              3.02918|\n",
      "|      SW Billings|              3.02917|\n",
      "|          V Kohli|              3.02917|\n",
      "|         AS Yadav|               3.0125|\n",
      "|        A Symonds|                  3.0|\n",
      "|   MJ McClenaghan|              2.99999|\n",
      "|       MJ Guptill|              2.98333|\n",
      "|         AR Patel|              2.97915|\n",
      "|         CL White|              2.95001|\n",
      "|         DJ Hooda|              2.94166|\n",
      "|        JP Duminy|              2.90833|\n",
      "|        RG Sharma|              2.88333|\n",
      "|            B Lee|              2.86668|\n",
      "|           P Negi|              2.84999|\n",
      "|          E Lewis|              2.84168|\n",
      "|         SE Marsh|              2.84166|\n",
      "|           N Rana|              2.83334|\n",
      "|         BJ Hodge|              2.83333|\n",
      "|        Q de Kock|              2.83333|\n",
      "|        ML Hayden|                2.825|\n",
      "|  Shakib Al Hasan|              2.82083|\n",
      "|       AD Mathews|              2.78751|\n",
      "|        RA Jadeja|               2.7875|\n",
      "|        DH Yagnik|              2.78334|\n",
      "|     F du Plessis|              2.78334|\n",
      "|        AT Rayudu|              2.73334|\n",
      "|    LA Pomersbach|              2.73333|\n",
      "|       RV Uthappa|              2.72917|\n",
      "|  Gurkeerat Singh|              2.72084|\n",
      "+-----------------+---------------------+\n",
      "only showing top 100 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "total_batting_weight = spark.sql('select hht.Batsman, round((hht.Weights+f.Weights+fsa.Weights+c.Weights+rbw.Weights), 5) as Total_Batting_Weights \\\n",
    "                                 from hard_hitting_ability hht \\\n",
    "                                 inner join finisher f \\\n",
    "                                 on hht.Batsman = f.Batsman \\\n",
    "                                 inner join fast_scoring_ability fsa \\\n",
    "                                 on hht.Batsman = fsa.Batsman \\\n",
    "                                 inner join consistency c \\\n",
    "                                 on hht.Batsman = c.Batsman \\\n",
    "                                 inner join running_between_wickets rbw \\\n",
    "                                 on hht.Batsman = rbw.Batsman \\\n",
    "                                 order by Total_Batting_Weights desc')\n",
    "total_batting_weight.registerTempTable('total_batting_weight')\n",
    "total_batting_weight.show(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------------+-----------+\n",
      "|database|           tableName|isTemporary|\n",
      "+--------+--------------------+-----------+\n",
      "|        |         analysis_db|       true|\n",
      "|        |       deliveries_db|       true|\n",
      "|        |          matches_db|       true|\n",
      "|        |total_batting_weight|       true|\n",
      "+--------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Dropping intermediate tables\n",
    "table_names = ['no_of_matches_table', 'hard_hitting_ability', 'finisher', 'fast_scoring_ability', 'consistency', 'running_between_wickets']\n",
    "for table in table_names:\n",
    "    cmd = 'drop table if exists {}'.format(table)\n",
    "    drop = spark.sql(cmd)\n",
    "check = spark.sql('show tables')\n",
    "check.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Bowling Metrics "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nmbo: no. of bowlers\n",
    "# nmb: no. of matches played by a bowler\n",
    "# eco: economy\n",
    "# wta: wicket taking ability\n",
    "# cons: consistency\n",
    "# cwta: crucial wicket taking ability\n",
    "# spi: short performance index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|No_of_Bowlers|\n",
      "+-------------+\n",
      "|          405|\n",
      "+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Number of Bowlers\n",
    "nmbo = spark.sql('Select count(distinct(bowler)) as No_of_Bowlers from analysis_db')\n",
    "nmbo.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of matches played by a bowler\n",
    "nmb = spark.sql('select bowler as Bowler, count(distinct(match_id)) as No_of_Matches from analysis_db group by bowler')\n",
    "nmb.registerTempTable('no_of_matches_bowlers')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Economy "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Economy = Runs Scored/(Number of balls bowled by bowler/6)\n",
    "eco = spark.sql('Select bowler as Bowler, round(runs/overs, 5) as Economy \\\n",
    "                from (Select bowler,round(count(*)/6) \\\n",
    "                as overs,sum(total_runs) as runs \\\n",
    "                from analysis_db \\\n",
    "                group by bowler)')\n",
    "eco.registerTempTable('economy')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "eco = spark.sql('select row_number() over (order by e.Economy asc) as Rank, e.*,n.No_of_Matches \\\n",
    "                from economy e \\\n",
    "                inner join \\\n",
    "                no_of_matches_bowlers n \\\n",
    "                on e.Bowler = n.Bowler where n.No_of_Matches>9')\n",
    "eco.registerTempTable('economy')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------------+-------+-------------+-------+\n",
      "|Rank|          Bowler|Economy|No_of_Matches|Weights|\n",
      "+----+----------------+-------+-------------+-------+\n",
      "|   1|   Sohail Tanvir|   6.25|           11|0.99528|\n",
      "|   2|      A Chandila|6.28205|           12|0.99057|\n",
      "|   3|         J Yadav|6.52632|           12|0.98585|\n",
      "|   4|      SM Pollock|6.53191|           13|0.98113|\n",
      "|   5|        A Kumble|6.64024|           42|0.97642|\n",
      "|   6|      GD McGrath|6.65455|           14| 0.9717|\n",
      "|   7|        DW Steyn|6.66848|           92|0.96698|\n",
      "|   8|  M Muralitharan|6.68561|           66|0.96226|\n",
      "|   9|RN ten Doeschate|6.71429|           10|0.95755|\n",
      "|  10|       RD Chahar|6.71429|           15|0.95283|\n",
      "+----+----------------+-------+-------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "eco = spark.sql('select *, round((212 - Rank)/212, 5) as Weights from economy')\n",
    "eco.registerTempTable('economy')\n",
    "eco.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Wicket Taking Ability"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Wicket Taking Ability = Number of balls bowled/Wickets Taken\n",
    "wta = spark.sql('(Select t1.bowler as Bowler, round(t2.balls/t1.wickets, 5) as Wicket_Taking_Ability from \\\n",
    "                (Select bowler,count(*) as wickets from analysis_db where player_dismissed is not null \\\n",
    "                and (dismissal_kind = \\'bowled\\' or  dismissal_kind = \\'hit wicket\\' \\\n",
    "                or  dismissal_kind = \\'stumped\\' or  dismissal_kind = \\'lbw\\' \\\n",
    "                or  dismissal_kind = \\'caught and bowled\\' or  dismissal_kind = \\'caught\\') \\\n",
    "                group by bowler) t1 \\\n",
    "                inner join \\\n",
    "                (select count(*) as balls,bowler from analysis_db group by bowler)t2 on \\\n",
    "                t1.bowler = t2.bowler)')\n",
    "wta.registerTempTable('wicket_taking_ability')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "wta = spark.sql('select row_number() over (order by w.Wicket_Taking_Ability asc) as Rank, w.*, n.No_of_Matches \\\n",
    "                from wicket_taking_ability w \\\n",
    "                inner join \\\n",
    "                no_of_matches_bowlers n on \\\n",
    "                w.Bowler = n.Bowler where n.No_of_Matches>9')\n",
    "wta.registerTempTable('wicket_taking_ability')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------+---------------------+-------------+-------+\n",
      "|Rank|        Bowler|Wicket_Taking_Ability|No_of_Matches|Weights|\n",
      "+----+--------------+---------------------+-------------+-------+\n",
      "|   1|       A Zampa|             11.84211|           11|0.99528|\n",
      "|   2| Sohail Tanvir|             12.04545|           11|0.99057|\n",
      "|   3|       K Ahmed|             12.68421|           10|0.98585|\n",
      "|   4|        N Rana|             13.42857|           12|0.98113|\n",
      "|   5|      K Rabada|                 14.0|           18|0.97642|\n",
      "|   6|      BJ Hodge|                 14.0|           20| 0.9717|\n",
      "|   7|  CRD Fernando|             14.64706|           10|0.96698|\n",
      "|   8|    YA Abdulla|                 14.8|           11|0.96226|\n",
      "|   9|A Ashish Reddy|                 15.0|           20|0.95755|\n",
      "|  10|       S Gopal|             15.60526|           30|0.95283|\n",
      "+----+--------------+---------------------+-------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "wta = spark.sql('select *,round((212-Rank)/212, 5) as Weights \\\n",
    "                from wicket_taking_ability')\n",
    "wta.registerTempTable('wicket_taking_ability')\n",
    "wta.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Consistency "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Consistency = Runs Conceded/Wickets Taken\n",
    "cons = spark.sql('select t1.bowler as Bowler, round(t1.runs/t2.wickets, 5) as Consistency \\\n",
    "                 from (select sum(total_runs) as runs,bowler from analysis_db group by bowler) t1 \\\n",
    "                 inner join \\\n",
    "                 (Select bowler,count(*) as wickets from analysis_db where player_dismissed is not null \\\n",
    "                 and (dismissal_kind = \\'bowled\\' or  dismissal_kind = \\'hit wicket\\' \\\n",
    "                 or dismissal_kind = \\'stumped\\' or  dismissal_kind = \\'lbw\\' \\\n",
    "                 or dismissal_kind = \\'caught and bowled\\' or  dismissal_kind = \\'caught\\') \\\n",
    "                 group by bowler)t2 on t1.bowler = t2.bowler')\n",
    "cons.registerTempTable('consistency')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "cons = spark.sql('select row_number() over (order by c.Consistency asc) as Rank, c.*,n.No_of_Matches \\\n",
    "                 from consistency c \\\n",
    "                 inner join \\\n",
    "                 no_of_matches_bowlers n on \\\n",
    "                 c.Bowler = n.Bowler where n.No_of_Matches>9')\n",
    "cons.registerTempTable('consistency')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------+-----------+-------------+-------+\n",
      "|Rank|        Bowler|Consistency|No_of_Matches|Weights|\n",
      "+----+--------------+-----------+-------------+-------+\n",
      "|   1| Sohail Tanvir|       12.5|           11|0.99528|\n",
      "|   2|       A Zampa|   14.78947|           11|0.99057|\n",
      "|   3|        N Rana|   17.71429|           12|0.98585|\n",
      "|   4|  CRD Fernando|       18.0|           10|0.98113|\n",
      "|   5|      BJ Hodge|   18.23529|           20|0.97642|\n",
      "|   6|       K Ahmed|   18.47368|           10| 0.9717|\n",
      "|   7|AD Mascarenhas|   19.21053|           13|0.96698|\n",
      "|   8|      K Rabada|   19.32258|           18|0.96226|\n",
      "|   9|  DE Bollinger|   19.35135|           27|0.95755|\n",
      "|  10|   MF Maharoof|    19.7037|           20|0.95283|\n",
      "+----+--------------+-----------+-------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cons = spark.sql('select *, round((212-Rank)/212, 5) as Weights from consistency')\n",
    "cons.registerTempTable('consistency')\n",
    "cons.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Crucial Wicket Taking Ability"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Crucial Wicket Taking Ability = Number of times Four or Five Wickets Taken/Number of Innings Played\n",
    "cwta = spark.sql('select t2.bowler as Bowler, round(nvl(t1.no_of_4wickets/t2.innings,0), 5) as Crucial_Wicket_Taking_Ablity \\\n",
    "                 from (select bowler,count(*) as no_of_4wickets from (select * from \\\n",
    "                 (select match_id,bowler,count(*) as wickets from analysis_db where player_dismissed \\\n",
    "                 is not null \\\n",
    "                 and (dismissal_kind = \\'bowled\\' or  dismissal_kind = \\'hit wicket\\' \\\n",
    "                 or  dismissal_kind = \\'stumped\\' or  dismissal_kind = \\'lbw\\' \\\n",
    "                 or  dismissal_kind = \\'caught and bowled\\' or  dismissal_kind = \\'caught\\') \\\n",
    "                 group by bowler,match_id ) \\\n",
    "                 where wickets > 3) group by bowler)t1 \\\n",
    "                 right join \\\n",
    "                 (select bowler,count(match_id) as \\\n",
    "                 innings from (select distinct(match_id),bowler from analysis_db) \\\n",
    "                 group by bowler)t2 \\\n",
    "                 on t1.bowler = t2.bowler order by Crucial_Wicket_Taking_Ablity desc')\n",
    "cwta.registerTempTable('crucial_wicket_taking_ablity')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "cwta = spark.sql('select rank() over (order by cw.Crucial_Wicket_Taking_Ablity desc) as Rank, cw.*,n.No_of_Matches \\\n",
    "                 from crucial_wicket_taking_ablity cw \\\n",
    "                 inner join no_of_matches_bowlers n on \\\n",
    "                 cw.Bowler = n.Bowler where n.No_of_Matches>9')\n",
    "cwta.registerTempTable('crucial_wicket_taking_ablity')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------+----------------------------+-------------+-------+\n",
      "|Rank|       Bowler|Crucial_Wicket_Taking_Ablity|No_of_Matches|Weights|\n",
      "+----+-------------+----------------------------+-------------+-------+\n",
      "|   1|Sohail Tanvir|                     0.18182|           11|0.99528|\n",
      "|   1|   YA Abdulla|                     0.18182|           11|0.99528|\n",
      "|   3|       AJ Tye|                     0.15385|           26|0.98585|\n",
      "|   4|     K Rabada|                     0.11111|           18|0.98113|\n",
      "|   5|     J Theron|                         0.1|           10|0.97642|\n",
      "|   5|  PC Valthaty|                         0.1|           10|0.97642|\n",
      "|   5| CRD Fernando|                         0.1|           10|0.97642|\n",
      "|   8|      A Zampa|                     0.09091|           11|0.96226|\n",
      "|   8|    CJ Jordan|                     0.09091|           11|0.96226|\n",
      "|  10|   A Chandila|                     0.08333|           12|0.95283|\n",
      "+----+-------------+----------------------------+-------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cwta = spark.sql('select *, round((212-Rank)/212, 5) as Weights \\\n",
    "                 from crucial_wicket_taking_ablity')\n",
    "cwta.registerTempTable('crucial_wicket_taking_ablity')\n",
    "cwta.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Short Performance Index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Short Performance Index = (Wickets Taken – 4* Number of Times Four Wickets Taken – 5* Number of Times Five Wickets Taken)/(Innings Played – Number of Times Four Wickets or Five Wickets Taken)\n",
    "spi = spark.sql('select n.bowler as Bowler, round(nvl(t5.Short_Performance_Index,0), 5) as Short_Performance_Index \\\n",
    "                from (select t1.bowler,(t3.wickets - 4*t1.no_of_4wickets - 5*t2.no_of_4wickets)/ \\\n",
    "                (t4.innings - t1.no_of_4wickets - t2.no_of_4wickets) as Short_Performance_Index \\\n",
    "                from (select bowler,count(*) as no_of_4wickets \\\n",
    "                from (select * from (select match_id,bowler,count(*) as wickets from analysis_db where player_dismissed \\\n",
    "                is not null \\\n",
    "                and (dismissal_kind = \\'bowled\\' or  dismissal_kind = \\'hit wicket\\' \\\n",
    "                or  dismissal_kind = \\'stumped\\' or  dismissal_kind = \\'lbw\\' \\\n",
    "                or  dismissal_kind = \\'caught and bowled\\' or  dismissal_kind = \\'caught\\')\\\n",
    "                group by bowler, match_id ) \\\n",
    "                where wickets = 4) group by bowler) t1 \\\n",
    "                inner join \\\n",
    "                (select bowler,count(*) as no_of_4wickets from (select * from \\\n",
    "                (select match_id,bowler,count(*) as wickets from analysis_db where player_dismissed \\\n",
    "                is not null \\\n",
    "                and (dismissal_kind = \\'bowled\\' or  dismissal_kind = \\'hit wicket\\' \\\n",
    "                or dismissal_kind = \\'stumped\\' or  dismissal_kind = \\'lbw\\' \\\n",
    "                or dismissal_kind = \\'caught and bowled\\' or  dismissal_kind = \\'caught\\')\\\n",
    "                group by bowler,match_id ) \\\n",
    "                where wickets = 5) group by bowler) t2 \\\n",
    "                inner join \\\n",
    "                (select bowler,count(*) as wickets from analysis_db where player_dismissed is not null \\\n",
    "                and (dismissal_kind = \\'bowled\\' or  dismissal_kind = \\'hit wicket\\' \\\n",
    "                or  dismissal_kind = \\'stumped\\' or  dismissal_kind = \\'lbw\\' \\\n",
    "                or  dismissal_kind = \\'caught and bowled\\' or  dismissal_kind = \\'caught\\') \\\n",
    "                group by bowler) t3 \\\n",
    "                inner join \\\n",
    "                (select bowler,count(match_id) as \\\n",
    "                innings from (select distinct(match_id),bowler from analysis_db) group by bowler) t4 \\\n",
    "                on t1.bowler = t2.bowler and t1.bowler = t3.bowler and t1.bowler = t4.bowler) t5 \\\n",
    "                right join \\\n",
    "                no_of_matches_bowlers n on t5.Bowler = n.Bowler order by Short_Performance_Index desc')\n",
    "spi.registerTempTable('short_performance_index')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "spi = spark.sql('select rank() over (order by sp.Short_Performance_Index desc) as Rank, sp.*, n.No_of_Matches \\\n",
    "                from short_performance_index sp \\\n",
    "                inner join \\\n",
    "                no_of_matches_bowlers n on \\\n",
    "                sp.Bowler = n.Bowler where n.No_of_Matches>9')\n",
    "spi.registerTempTable('short_performance_index')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---------------+-----------------------+-------------+-------+\n",
      "|Rank|         Bowler|Short_Performance_Index|No_of_Matches|Weights|\n",
      "+----+---------------+-----------------------+-------------+-------+\n",
      "|   1|     SL Malinga|                1.22609|          122|0.99528|\n",
      "|   2|        B Kumar|                1.05263|          117|0.99057|\n",
      "|   3|       MM Patel|                1.01667|           63|0.98585|\n",
      "|   4|         AJ Tye|                    1.0|           26|0.98113|\n",
      "|   5|       A Mishra|                0.97203|          147|0.97642|\n",
      "|   6|      SP Narine|                0.91176|          109| 0.9717|\n",
      "|   7|Harbhajan Singh|                0.90968|          157|0.96698|\n",
      "|   8|       L Balaji|                0.85507|           73|0.96226|\n",
      "|   9|    JP Faulkner|                0.82456|           60|0.95755|\n",
      "|  10|       A Kumble|                0.82051|           42|0.95283|\n",
      "+----+---------------+-----------------------+-------------+-------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spi = spark.sql('select *, round((212-Rank)/212, 5) as Weights \\\n",
    "                from short_performance_index')\n",
    "spi.registerTempTable('short_performance_index')\n",
    "spi.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Table Name for each Metric\n",
    "# Economy: economy\n",
    "# Wicket Taking Ability: wicket_taking_ability\n",
    "# Consistency: consistency\n",
    "# Crucial Wicket Taking Ability: crucial_wicket_taking_ablity\n",
    "# Short Performance Index: short_performance_index"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Total Bowling Weights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+---------------------+\n",
      "|         Bowler|Total_Bowling_Weights|\n",
      "+---------------+---------------------+\n",
      "|  Sohail Tanvir|              4.91981|\n",
      "|   CRD Fernando|              4.68397|\n",
      "| AD Mascarenhas|              4.67925|\n",
      "|        A Zampa|              4.67925|\n",
      "|     SL Malinga|              4.59906|\n",
      "|   DE Bollinger|              4.51416|\n",
      "|     A Chandila|              4.45755|\n",
      "|       MA Starc|              4.42454|\n",
      "|NM Coulter-Nile|              4.34435|\n",
      "|      SP Narine|              4.33491|\n",
      "+---------------+---------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "total_bowling_weight = spark.sql('select e.bowler as Bowler, round((e.Weights+wta.Weights+c.Weights+cwta.Weights+spi.Weights), 5) as Total_Bowling_Weights \\\n",
    "                                 from economy e \\\n",
    "                                 inner join wicket_taking_ability wta \\\n",
    "                                 on e.Bowler = wta.Bowler \\\n",
    "                                 inner join consistency c \\\n",
    "                                 on e.Bowler = c.Bowler \\\n",
    "                                 inner join crucial_wicket_taking_ablity cwta \\\n",
    "                                 on e.Bowler = cwta.Bowler \\\n",
    "                                 inner join short_performance_index spi \\\n",
    "                                 on e.Bowler = spi.Bowler \\\n",
    "                                 order by Total_Bowling_Weights desc')\n",
    "total_bowling_weight.registerTempTable('total_bowling_weight')\n",
    "total_bowling_weight.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------------+-----------+\n",
      "|database|           tableName|isTemporary|\n",
      "+--------+--------------------+-----------+\n",
      "|        |         analysis_db|       true|\n",
      "|        |       deliveries_db|       true|\n",
      "|        |          matches_db|       true|\n",
      "|        |total_batting_weight|       true|\n",
      "|        |total_bowling_weight|       true|\n",
      "+--------+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Dropping intermediate Tables\n",
    "table_names = ['no_of_matches_bowlers', 'economy', 'wicket_taking_ability', 'consistency', 'crucial_wicket_taking_ablity', 'short_performance_index']\n",
    "for table in table_names:\n",
    "    cmd = 'drop table if exists {}'.format(table)\n",
    "    drop = spark.sql(cmd)\n",
    "check = spark.sql('show tables')\n",
    "check.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Total Weights per Player"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+--------------------+--------------------+------------+\n",
      "|           Player|Total_Batting_Weight|Total_Bowling_Weight|Total_Weight|\n",
      "+-----------------+--------------------+--------------------+------------+\n",
      "|        CH Morris|             4.60416|             3.74057|     8.34473|\n",
      "|      MF Maharoof|             4.02916|             4.26415|     8.29331|\n",
      "|      Rashid Khan|             4.14583|             4.12265|     8.26848|\n",
      "|    Mohammad Nabi|             4.12916|             3.77831|     7.90747|\n",
      "|        KK Cooper|             3.88334|              3.9953|     7.87864|\n",
      "|        SP Narine|             3.53749|             4.33491|      7.8724|\n",
      "|           AJ Tye|             3.69167|             4.08491|     7.77658|\n",
      "|   A Ashish Reddy|               4.125|             3.51887|     7.64387|\n",
      "|         M Morkel|             4.04165|             3.58491|     7.62656|\n",
      "|Washington Sundar|             4.38332|              3.2217|     7.60502|\n",
      "+-----------------+--------------------+--------------------+------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "total_weight = spark.sql('select *, round((coalesce(total_batting_weight, 0) + coalesce(total_bowling_weight, 0)),5) as Total_Weight \\\n",
    "                          from (select t1.batsman as Player, round(nvl(t1.Total_Batting_Weights,0),5) as Total_Batting_Weight, \\\n",
    "                          round(nvl(t2.Total_Bowling_Weights,0),5) as Total_Bowling_Weight \\\n",
    "                          from total_batting_weight t1 \\\n",
    "                          full outer join total_bowling_weight t2 \\\n",
    "                          on t1.batsman = t2.bowler) \\\n",
    "                          order by Total_Weight desc')\n",
    "total_weight.registerTempTable('total_weight')\n",
    "total_weight.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------+-----------+\n",
      "|database|    tableName|isTemporary|\n",
      "+--------+-------------+-----------+\n",
      "|        |  analysis_db|       true|\n",
      "|        |deliveries_db|       true|\n",
      "|        |   matches_db|       true|\n",
      "|        | total_weight|       true|\n",
      "+--------+-------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Dropping intermediate tables\n",
    "table_names = ['total_batting_weight', 'total_bowling_weight']\n",
    "for table in table_names:\n",
    "    cmd = 'drop table if exists {}'.format(table)\n",
    "    drop = spark.sql(cmd)\n",
    "check = spark.sql('show tables')\n",
    "check.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Saving the player weight data\n",
    "total_weight.toPandas().to_csv(path + '/dataset/weights_data/player_weights.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "'Python Interactive'",
   "language": "python",
   "name": "d0f17b5a-28c8-4865-a251-18f09addc1f1"
  },
  "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.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}