IPL_Data_Preparation.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "collapsed": true,
     "inputWidgets": {},
     "nuid": "2c1c2bfc-8799-4864-bfb1-0d226f45c733",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"/?o=3997074275245566#setting/sparkui/1013-132650-t8top3kx/driver-6724931370309490260\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.3.2</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[8]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Databricks Shell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ]
     },
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "addedWidgets": {},
       "arguments": {},
       "data": "\n            <div>\n                <p><b>SparkSession - hive</b></p>\n                \n        <div>\n            <p><b>SparkContext</b></p>\n\n            <p><a href=\"/?o=3997074275245566#setting/sparkui/1013-132650-t8top3kx/driver-6724931370309490260\">Spark UI</a></p>\n\n            <dl>\n              <dt>Version</dt>\n                <dd><code>v3.3.2</code></dd>\n              <dt>Master</dt>\n                <dd><code>local[8]</code></dd>\n              <dt>AppName</dt>\n                <dd><code>Databricks Shell</code></dd>\n            </dl>\n        </div>\n        \n            </div>\n        ",
       "datasetInfos": [],
       "metadata": {},
       "removedWidgets": [],
       "textData": null,
       "type": "htmlSandbox"
      }
     },
     "output_type": "display_data"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9747c593-4fa0-4b40-8c78-8b7878520b12",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructField, StructType, IntegerType, StringType, BooleanType, DateType, DecimalType\n",
    "from pyspark.sql import Window, functions as F\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "collapsed": true,
     "inputWidgets": {},
     "nuid": "904ba1b5-3c64-4639-8da2-8f3d9a3bf8ec",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "#create session\n",
    "spark = SparkSession.builder.appName(\"IPL Data Analysis\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "collapsed": true,
     "inputWidgets": {},
     "nuid": "6b9f2855-30ad-4913-94af-7e35a581e40d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"/?o=3997074275245566#setting/sparkui/1013-132650-t8top3kx/driver-6724931370309490260\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.3.2</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[8]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Databricks Shell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ]
     },
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "addedWidgets": {},
       "arguments": {},
       "data": "\n            <div>\n                <p><b>SparkSession - hive</b></p>\n                \n        <div>\n            <p><b>SparkContext</b></p>\n\n            <p><a href=\"/?o=3997074275245566#setting/sparkui/1013-132650-t8top3kx/driver-6724931370309490260\">Spark UI</a></p>\n\n            <dl>\n              <dt>Version</dt>\n                <dd><code>v3.3.2</code></dd>\n              <dt>Master</dt>\n                <dd><code>local[8]</code></dd>\n              <dt>AppName</dt>\n                <dd><code>Databricks Shell</code></dd>\n            </dl>\n        </div>\n        \n            </div>\n        ",
       "datasetInfos": [],
       "metadata": {},
       "removedWidgets": [],
       "textData": null,
       "type": "htmlSandbox"
      }
     },
     "output_type": "display_data"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "04346e6e-5860-4bac-b35a-6667953f52c0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n |-- MatcH_id: string (nullable = true)\n |-- Over_id: string (nullable = true)\n |-- Ball_id: string (nullable = true)\n |-- Innings_No: string (nullable = true)\n |-- Team_Batting: string (nullable = true)\n |-- Team_Bowling: string (nullable = true)\n |-- Striker_Batting_Position: string (nullable = true)\n |-- Extra_Type: string (nullable = true)\n |-- Runs_Scored: string (nullable = true)\n |-- Extra_runs: string (nullable = true)\n |-- Wides: string (nullable = true)\n |-- Legbyes: string (nullable = true)\n |-- Byes: string (nullable = true)\n |-- Noballs: string (nullable = true)\n |-- Penalty: string (nullable = true)\n |-- Bowler_Extras: string (nullable = true)\n |-- Out_type: string (nullable = true)\n |-- Caught: string (nullable = true)\n |-- Bowled: string (nullable = true)\n |-- Run_out: string (nullable = true)\n |-- LBW: string (nullable = true)\n |-- Retired_hurt: string (nullable = true)\n |-- Stumped: string (nullable = true)\n |-- caught_and_bowled: string (nullable = true)\n |-- hit_wicket: string (nullable = true)\n |-- ObstructingFeild: string (nullable = true)\n |-- Bowler_Wicket: string (nullable = true)\n |-- Match_Date: string (nullable = true)\n |-- Season: string (nullable = true)\n |-- Striker: string (nullable = true)\n |-- Non_Striker: string (nullable = true)\n |-- Bowler: string (nullable = true)\n |-- Player_Out: string (nullable = true)\n |-- Fielders: string (nullable = true)\n |-- Striker_match_SK: string (nullable = true)\n |-- StrikerSK: string (nullable = true)\n |-- NonStriker_match_SK: string (nullable = true)\n |-- NONStriker_SK: string (nullable = true)\n |-- Fielder_match_SK: string (nullable = true)\n |-- Fielder_SK: string (nullable = true)\n |-- Bowler_match_SK: string (nullable = true)\n |-- BOWLER_SK: string (nullable = true)\n |-- PlayerOut_match_SK: string (nullable = true)\n |-- BattingTeam_SK: string (nullable = true)\n |-- BowlingTeam_SK: string (nullable = true)\n |-- Keeper_Catch: string (nullable = true)\n |-- Player_out_sk: string (nullable = true)\n |-- MatchDateSK: string (nullable = true)\n\n"
     ]
    }
   ],
   "source": [
    "ball_by_ball_df_raw = spark.read.format(\"csv\").option(\"header\", \"true\").load(\"s3://ipl-data-analysis-project/Ball_By_Ball.csv\")\n",
    "ball_by_ball_df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d92a5526-856d-456f-95b2-336c4654efb9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "ball_by_ball_df = ball_by_ball_df_raw \\\n",
    "    .withColumn(\"match_id\", F.col(\"MatcH_id\").cast(IntegerType())) \\\n",
    "    .withColumn(\"over_id\", F.col(\"Over_id\").cast(IntegerType())) \\\n",
    "    .withColumn(\"ball_id\", F.col(\"Ball_id\").cast(IntegerType())) \\\n",
    "    .withColumn(\"innings_no\", F.col(\"Innings_No\").cast(IntegerType())) \\\n",
    "    .withColumn(\"team_batting\", F.col(\"Team_Batting\").cast(StringType())) \\\n",
    "    .withColumn(\"team_bowling\", F.col(\"Team_Bowling\").cast(StringType())) \\\n",
    "    .withColumn(\"striker_batting_position\", F.col(\"Striker_Batting_Position\").cast(IntegerType())) \\\n",
    "    .withColumn(\"extra_type\", F.col(\"Extra_Type\").cast(StringType())) \\\n",
    "    .withColumn(\"runs_scored\", F.col(\"Runs_Scored\").cast(IntegerType())) \\\n",
    "    .withColumn(\"extra_runs\", F.col(\"Extra_runs\").cast(IntegerType())) \\\n",
    "    .withColumn(\"wides\", F.col(\"Wides\").cast(IntegerType())) \\\n",
    "    .withColumn(\"legbyes\", F.col(\"Legbyes\").cast(IntegerType())) \\\n",
    "    .withColumn(\"byes\", F.col(\"Byes\").cast(IntegerType())) \\\n",
    "    .withColumn(\"noballs\", F.col(\"Noballs\").cast(IntegerType())) \\\n",
    "    .withColumn(\"penalty\", F.col(\"Penalty\").cast(IntegerType())) \\\n",
    "    .withColumn(\"bowler_extras\", F.col(\"Bowler_Extras\").cast(IntegerType())) \\\n",
    "    .withColumn(\"out_type\", F.col(\"Out_type\").cast(StringType())) \\\n",
    "    .withColumn(\"caught\", F.col(\"Caught\").cast(BooleanType())) \\\n",
    "    .withColumn(\"bowled\", F.col(\"Bowled\").cast(BooleanType())) \\\n",
    "    .withColumn(\"run_out\", F.col(\"Run_out\").cast(BooleanType())) \\\n",
    "    .withColumn(\"lbw\", F.col(\"LBW\").cast(BooleanType())) \\\n",
    "    .withColumn(\"retired_hurt\", F.col(\"Retired_hurt\").cast(BooleanType())) \\\n",
    "    .withColumn(\"stumped\", F.col(\"Stumped\").cast(BooleanType())) \\\n",
    "    .withColumn(\"caught_and_bowled\", F.col(\"caught_and_bowled\").cast(BooleanType())) \\\n",
    "    .withColumn(\"hit_wicket\", F.col(\"hit_wicket\").cast(BooleanType())) \\\n",
    "    .withColumn(\"obstructingfeild\", F.col(\"ObstructingFeild\").cast(BooleanType())) \\\n",
    "    .withColumn(\"bowler_wicket\", F.col(\"Bowler_Wicket\").cast(BooleanType())) \\\n",
    "    .withColumn(\"match_date\", F.to_date(F.col(\"match_date\"), \"M/d/yyyy\")) \\\n",
    "    .withColumn(\"season\", F.col(\"Season\").cast(IntegerType())) \\\n",
    "    .withColumn(\"striker\", F.col(\"Striker\").cast(IntegerType())) \\\n",
    "    .withColumn(\"non_striker\", F.col(\"Non_Striker\").cast(IntegerType())) \\\n",
    "    .withColumn(\"bowler\", F.col(\"Bowler\").cast(IntegerType())) \\\n",
    "    .withColumn(\"player_out\", F.col(\"Player_Out\").cast(IntegerType())) \\\n",
    "    .withColumn(\"fielders\", F.col(\"Fielders\").cast(IntegerType())) \\\n",
    "    .withColumn(\"striker_match_sk\", F.col(\"Striker_match_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"strikersk\", F.col(\"StrikerSK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"nonstriker_match_sk\", F.col(\"NonStriker_match_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"nonstriker_sk\", F.col(\"NONStriker_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"fielder_match_sk\", F.col(\"Fielder_match_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"fielder_sk\", F.col(\"Fielder_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"bowler_match_sk\", F.col(\"Bowler_match_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"bowler_sk\", F.col(\"BOWLER_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"playerout_match_sk\", F.col(\"PlayerOut_match_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"battingteam_sk\", F.col(\"BattingTeam_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"bowlingteam_sk\", F.col(\"BowlingTeam_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"keeper_catch\", F.col(\"Keeper_Catch\").cast(BooleanType())) \\\n",
    "    .withColumn(\"player_out_sk\", F.col(\"Player_out_sk\").cast(IntegerType())) \\\n",
    "    .withColumn(\"matchdatesk\", F.to_date(F.col(\"MatchDateSK\"), \"yyyyMMdd\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b07f7c9e-50c1-44f6-bed5-e46b16bc6d96",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------+-------+----------+------------+------------+------------------------+----------+-----------+----------+-----+-------+----+-------+-------+-------------+--------------+------+------+-------+-----+------------+-------+-----------------+----------+----------------+-------------+----------+------+-------+-----------+------+----------+--------+----------------+---------+-------------------+-------------+----------------+----------+---------------+---------+------------------+--------------+--------------+------------+-------------+-----------+\n|match_id|over_id|ball_id|innings_no|team_batting|team_bowling|striker_batting_position|extra_type|runs_scored|extra_runs|wides|legbyes|byes|noballs|penalty|bowler_extras|      out_type|caught|bowled|run_out|  lbw|retired_hurt|stumped|caught_and_bowled|hit_wicket|obstructingfeild|bowler_wicket|match_date|season|striker|non_striker|bowler|player_out|fielders|striker_match_sk|strikersk|nonstriker_match_sk|nonstriker_sk|fielder_match_sk|fielder_sk|bowler_match_sk|bowler_sk|playerout_match_sk|battingteam_sk|bowlingteam_sk|keeper_catch|player_out_sk|matchdatesk|\n+--------+-------+-------+----------+------------+------------+------------------------+----------+-----------+----------+-----+-------+----+-------+-------+-------------+--------------+------+------+-------+-----+------------+-------+-----------------+----------+----------------+-------------+----------+------+-------+-----------+------+----------+--------+----------------+---------+-------------------+-------------+----------------+----------+---------------+---------+------------------+--------------+--------------+------------+-------------+-----------+\n|  598028|     15|      6|         1|           5|           2|                       6| No Extras|          4|         0|    0|      0|   0|      0|      0|            0|Not Applicable| false| false|  false|false|       false|  false|            false|     false|           false|        false|2013-04-20|  2013|    277|        104|    83|      null|    null|           20336|      276|              20333|          103|              -1|        -1|          20343|       82|                -1|             4|             1|       false|            0| 2013-04-20|\n+--------+-------+-------+----------+------------+------------+------------------------+----------+-----------+----------+-----+-------+----+-------+-------+-------------+--------------+------+------+-------+-----+------------+-------+-----------------+----------+----------------+-------------+----------+------+-------+-----------+------+----------+--------+----------------+---------+-------------------+-------------+----------------+----------+---------------+---------+------------------+--------------+--------------+------------+-------------+-----------+\nonly showing top 1 row\n\n"
     ]
    }
   ],
   "source": [
    "ball_by_ball_df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "95d71b8f-36b5-49c0-844f-0d5bc297f9ec",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n |-- Match_SK: string (nullable = true)\n |-- match_id: string (nullable = true)\n |-- Team1: string (nullable = true)\n |-- Team2: string (nullable = true)\n |-- match_date: string (nullable = true)\n |-- Season_Year: string (nullable = true)\n |-- Venue_Name: string (nullable = true)\n |-- City_Name: string (nullable = true)\n |-- Country_Name: string (nullable = true)\n |-- Toss_Winner: string (nullable = true)\n |-- match_winner: string (nullable = true)\n |-- Toss_Name: string (nullable = true)\n |-- Win_Type: string (nullable = true)\n |-- Outcome_Type: string (nullable = true)\n |-- ManOfMach: string (nullable = true)\n |-- Win_Margin: string (nullable = true)\n |-- Country_id: string (nullable = true)\n\n"
     ]
    }
   ],
   "source": [
    "match_df_raw = spark.read.format(\"csv\").option(\"header\",\"true\").load(\"s3://ipl-data-analysis-project/Match.csv\")\n",
    "match_df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6d7778b4-7c90-4f5e-ad64-de86612e03e6",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "match_df = match_df_raw \\\n",
    "    .withColumn(\"match_sk\", F.col(\"Match_SK\").cast(IntegerType())) \\\n",
    "    .withColumn(\"match_id\", F.col(\"match_id\").cast(IntegerType())) \\\n",
    "    .withColumn(\"match_date\", F.to_date(\"match_date\",\"M/d/yyyy\")) \\\n",
    "    .withColumn(\"season_year\", F.col(\"season_year\").cast(IntegerType())) \\\n",
    "    .withColumn(\"win_margin\", F.col(\"win_margin\").cast(IntegerType())) \\\n",
    "    .withColumn(\"country_id\", F.col(\"country_id\").cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f306695a-3e44-43ac-a91b-5b9f00dbad5b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------+--------------------+--------------------+----------+-----------+--------------------+---------+------------+--------------------+--------------------+---------+--------+------------+-----------+----------+----------+\n|match_sk|match_id|               Team1|               Team2|match_date|season_year|          Venue_Name|City_Name|Country_Name|         Toss_Winner|        match_winner|Toss_Name|Win_Type|Outcome_Type|  ManOfMach|win_margin|country_id|\n+--------+--------+--------------------+--------------------+----------+-----------+--------------------+---------+------------+--------------------+--------------------+---------+--------+------------+-----------+----------+----------+\n|       0|  335987|Royal Challengers...|Kolkata Knight Ri...|2008-04-18|       2008|M Chinnaswamy Sta...|Bangalore|       India|Royal Challengers...|Kolkata Knight Ri...|    field|    runs|      Result|BB McCullum|       140|         1|\n+--------+--------+--------------------+--------------------+----------+-----------+--------------------+---------+------------+--------------------+--------------------+---------+--------+------------+-----------+----------+----------+\nonly showing top 1 row\n\n"
     ]
    }
   ],
   "source": [
    "match_df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5031ee69-09af-4153-bcad-f3f5884bcc0b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n |-- PLAYER_SK: string (nullable = true)\n |-- Player_Id: string (nullable = true)\n |-- Player_Name: string (nullable = true)\n |-- DOB: string (nullable = true)\n |-- Batting_hand: string (nullable = true)\n |-- Bowling_skill: string (nullable = true)\n |-- Country_Name: string (nullable = true)\n\n"
     ]
    }
   ],
   "source": [
    "player_df_raw = spark.read.format(\"csv\").option(\"header\",\"true\").option(\"dateFormat\",\"M/d/yyyy\").load(\"s3://ipl-data-analysis-project/Player.csv\")\n",
    "player_df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "ba819493-f4e1-4226-bfe0-fdb4ce9626e3",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "player_df = player_df_raw \\\n",
    "    .withColumn(\"player_sk\", F.col(\"player_sk\").cast(IntegerType())) \\\n",
    "    .withColumn(\"player_id\", F.col(\"player_id\").cast(IntegerType())) \\\n",
    "    .withColumn(\"dob\", F.to_date(\"dob\", \"M/d/yyyy\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "921940c4-5d40-41e6-a9bd-63cdf0653cab",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----------+----------+-------------+----------------+------------+\n|player_sk|player_id|Player_Name|       dob| Batting_hand|   Bowling_skill|Country_Name|\n+---------+---------+-----------+----------+-------------+----------------+------------+\n|        0|        1| SC Ganguly|1972-07-08|Left-hand bat|Right-arm medium|       India|\n+---------+---------+-----------+----------+-------------+----------------+------------+\nonly showing top 1 row\n\n"
     ]
    }
   ],
   "source": [
    "player_df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "89c778bc-348f-4e29-a7c9-62ebaba486b4",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n |-- Player_match_SK: string (nullable = true)\n |-- PlayerMatch_key: string (nullable = true)\n |-- Match_Id: string (nullable = true)\n |-- Player_Id: string (nullable = true)\n |-- Player_Name: string (nullable = true)\n |-- DOB: string (nullable = true)\n |-- Batting_hand: string (nullable = true)\n |-- Bowling_skill: string (nullable = true)\n |-- Country_Name: string (nullable = true)\n |-- Role_Desc: string (nullable = true)\n |-- Player_team: string (nullable = true)\n |-- Opposit_Team: string (nullable = true)\n |-- Season_year: string (nullable = true)\n |-- is_manofThematch: string (nullable = true)\n |-- Age_As_on_match: string (nullable = true)\n |-- IsPlayers_Team_won: string (nullable = true)\n |-- Batting_Status: string (nullable = true)\n |-- Bowling_Status: string (nullable = true)\n |-- Player_Captain: string (nullable = true)\n |-- Opposit_captain: string (nullable = true)\n |-- Player_keeper: string (nullable = true)\n |-- Opposit_keeper: string (nullable = true)\n\n"
     ]
    }
   ],
   "source": [
    "player_match_df_raw = spark.read.format(\"csv\").option(\"header\",\"true\").load(\"s3://ipl-data-analysis-project/Player_match.csv\")\n",
    "player_match_df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "dfd17db9-a09b-4963-b72c-e787d51a2696",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "player_match_df = player_match_df_raw \\\n",
    "    .withColumn(\"player_match_sk\", F.col(\"player_match_sk\").cast(IntegerType())) \\\n",
    "    .withColumn(\"playermatch_key\", F.col(\"playermatch_key\").cast(DecimalType())) \\\n",
    "    .withColumn(\"match_id\", F.col(\"match_id\").cast(IntegerType())) \\\n",
    "    .withColumn(\"player_id\", F.col(\"player_id\").cast(IntegerType())) \\\n",
    "    .withColumn(\"dob\", F.to_date(\"dob\",\"M/d/yyyy\")) \\\n",
    "    .withColumn(\"season_year\", F.col(\"season_year\").cast(IntegerType())) \\\n",
    "    .withColumn(\"age_as_on_match\", F.col(\"age_as_on_match\").cast(IntegerType())) \\\n",
    "    .withColumn(\"is_manofthematch\", F.col(\"is_manofthematch\").cast(BooleanType())) \\\n",
    "    .withColumn(\"isplayers_team_won\", F.col(\"isplayers_team_won\").cast(BooleanType())) \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2b619f24-c935-4ee9-8ab0-5bba0236190e",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+---------------+--------+---------+-----------+----------+--------------+--------------------+------------+---------+--------------------+--------------------+-----------+----------------+---------------+------------------+--------------+--------------+--------------+---------------+-------------+--------------+\n|player_match_sk|playermatch_key|match_id|player_id|Player_Name|       dob|  Batting_hand|       Bowling_skill|Country_Name|Role_Desc|         Player_team|        Opposit_Team|season_year|is_manofthematch|age_as_on_match|isplayers_team_won|Batting_Status|Bowling_Status|Player_Captain|Opposit_captain|Player_keeper|Opposit_keeper|\n+---------------+---------------+--------+---------+-----------+----------+--------------+--------------------+------------+---------+--------------------+--------------------+-----------+----------------+---------------+------------------+--------------+--------------+--------------+---------------+-------------+--------------+\n|             -1|             -1|      -1|       -1|        N/A|      null|          null|                null|        null|     null|                null|                null|       null|            null|           null|              null|          null|          null|          null|           null|         null|          null|\n|          12694|           null|  335987|        6|   R Dravid|1973-01-11|Right-hand bat|  Right-arm offbreak|       India|  Captain|Royal Challengers...|Kolkata Knight Ri...|       2008|           false|             35|             false|          null|          null|      R Dravid|     SC Ganguly|   MV Boucher|       WP Saha|\n|          12695|           null|  335987|        7|   W Jaffer|1978-02-16|Right-hand bat|  Right-arm offbreak|       India|   Player|Royal Challengers...|Kolkata Knight Ri...|       2008|           false|             30|             false|          null|          null|      R Dravid|     SC Ganguly|   MV Boucher|       WP Saha|\n|          12696|           null|  335987|        8|    V Kohli|1988-11-05|Right-hand bat|    Right-arm medium|       India|   Player|Royal Challengers...|Kolkata Knight Ri...|       2008|           false|             20|             false|          null|          null|      R Dravid|     SC Ganguly|   MV Boucher|       WP Saha|\n|          12697|           null|  335987|        9|  JH Kallis|1975-10-16|Right-hand bat|Right-arm fast-me...|South Africa|   Player|Royal Challengers...|Kolkata Knight Ri...|       2008|           false|             33|             false|          null|          null|      R Dravid|     SC Ganguly|   MV Boucher|       WP Saha|\n+---------------+---------------+--------+---------+-----------+----------+--------------+--------------------+------------+---------+--------------------+--------------------+-----------+----------------+---------------+------------------+--------------+--------------+--------------+---------------+-------------+--------------+\nonly showing top 5 rows\n\n"
     ]
    }
   ],
   "source": [
    "player_match_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b510b204-a97a-4f73-9496-36a2fe5ee945",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n |-- Team_SK: string (nullable = true)\n |-- Team_Id: string (nullable = true)\n |-- Team_Name: string (nullable = true)\n\n"
     ]
    }
   ],
   "source": [
    "team_df_raw = spark.read.format(\"csv\").option(\"header\",\"true\").load(\"s3://ipl-data-analysis-project/Team.csv\")\n",
    "team_df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b3136ea9-fb9c-4d5a-bb88-84777262845c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "team_df = team_df_raw \\\n",
    "    .withColumn(\"team_sk\", F.col(\"team_sk\").cast(IntegerType())) \\\n",
    "    .withColumn(\"team_id\", F.col(\"team_id\").cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "fed1a864-947c-4c0c-b6b0-703ce6469c76",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+--------------------+\n|team_sk|team_id|           Team_Name|\n+-------+-------+--------------------+\n|      0|      1|Kolkata Knight Ri...|\n|      1|      2|Royal Challengers...|\n|      2|      3| Chennai Super Kings|\n|      3|      4|     Kings XI Punjab|\n|      4|      5|    Rajasthan Royals|\n+-------+-------+--------------------+\nonly showing top 5 rows\n\n"
     ]
    }
   ],
   "source": [
    "team_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "40d733b5-cebf-4c7c-a2b4-0b9a7171b417",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# No of bowler extras by each team per season\n",
    "\n",
    "joined_df = ball_by_ball_df.join(team_df, ball_by_ball_df[\"team_bowling\"] == team_df[\"team_id\"], \"inner\")\n",
    "grouped_df = joined_df.groupBy(\"team_name\", \"season\").sum(\"bowler_extras\").withColumnRenamed(\"sum(bowler_extras)\", \"total_bowler_extras\")\n",
    "sorted_df = grouped_df.orderBy(grouped_df.season.asc(), grouped_df.total_bowler_extras.desc())\n",
    "output_path = \"dbfs:/FileStore/tables/bowler_extras_by_team_per_season\"\n",
    "sorted_df.write.csv(output_path, header=True, mode=\"overwrite\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d5c84000-e6d5-4084-b467-794513dc8439",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# No of extra runs by each team per season\n",
    "\n",
    "joined_df = ball_by_ball_df.join(team_df, ball_by_ball_df[\"team_batting\"] == team_df[\"team_id\"], \"inner\")\n",
    "grouped_df = joined_df.groupBy(\"team_name\", \"season\").sum(\"extra_runs\").withColumnRenamed(\"sum(extra_runs)\", \"total_extra_runs\")\n",
    "sorted_df = grouped_df.orderBy(grouped_df.season.asc(), grouped_df.total_extra_runs.desc())\n",
    "output_path = \"dbfs:/FileStore/tables/extras_by_team_per_season\"\n",
    "sorted_df.write.csv(output_path, header=True, mode=\"overwrite\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4108b55f-f1b7-4216-8dc3-2c25cddb9dab",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# No of extras by top 5 bowlers each season\n",
    "\n",
    "grouped_df = ball_by_ball_df.groupBy(\"bowler\", \"season\").sum(\"bowler_extras\").withColumnRenamed(\"sum(bowler_extras)\", \"total_bowler_extras\")\n",
    "joined_df = grouped_df.join(player_df, grouped_df[\"bowler\"] == player_df[\"player_id\"], \"inner\")\n",
    "select_df = joined_df.select(\"player_name\", \"country_name\", \"season\", \"total_bowler_extras\")\n",
    "output_path = \"dbfs:/FileStore/tables/top_bowlers_with_most_extras\"\n",
    "select_df.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "04644865-9634-4cac-9df4-0a5579026aae",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# No of sixes by top 5 batters each season\n",
    "\n",
    "filter_sixes_df = ball_by_ball_df.filter(F.col(\"runs_scored\") == 6)\n",
    "grouped_df = filter_sixes_df.groupBy(\"striker\", \"season\").count().withColumnRenamed(\"count\", \"total_sixes\")\n",
    "joined_df = grouped_df.join(player_df, grouped_df[\"striker\"] == player_df[\"player_id\"] , \"inner\")\n",
    "select_df = joined_df.select(\"player_name\", \"country_name\", \"season\", \"total_sixes\")\n",
    "output_path = \"dbfs:/FileStore/tables/top_batters_with_most_runs\"\n",
    "select_df.write.csv(output_path, header=True, mode=\"overwrite\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "cf171826-f49a-4d95-a953-281e72843577",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Determine which players have the best batting averages over the years.\n",
    "\n",
    "# Step 1: Calculate Total Outs by Each Player in Each Season\n",
    "# Filter for deliveries that resulted in an out, group by player and season, and count the total outs.\n",
    "player_outs_per_season_df = ball_by_ball_df.select(\"striker\", \"season\") \\\n",
    "    .where(F.col(\"bowler_wicket\") == 'true') \\\n",
    "    .groupBy(\"striker\", \"season\") \\\n",
    "    .agg(F.count(\"striker\").alias(\"total_outs\"))\n",
    "\n",
    "# Step 2: Calculate Total Runs Scored by Each Player in Each Season\n",
    "# Group by player and season, and sum the runs scored.\n",
    "player_runs_per_season_df = ball_by_ball_df.select(\"striker\", \"season\", \"runs_scored\") \\\n",
    "    .groupBy(\"striker\", \"season\") \\\n",
    "    .agg(F.sum(\"runs_scored\").alias(\"total_runs\"))\n",
    "\n",
    "# Step 3: Join Runs and Outs Data for Each Player per Season\n",
    "# Join the runs and outs data to have both total runs and outs for each player-season.\n",
    "seasonal_runs_outs_df = player_runs_per_season_df \\\n",
    "    .join(player_outs_per_season_df, [\"striker\", \"season\"], \"inner\") \\\n",
    "    .select(\"striker\", \"season\", \"total_runs\", \"total_outs\")\n",
    "\n",
    "# Step 4: Define a Window Specification for Calculating Cumulative Sums per Player per Season\n",
    "# Partition by player to calculate cumulative statistics within each player group, ordered by season.\n",
    "player_season_window_spec = Window.partitionBy(\"striker\").orderBy(\"season\")\n",
    "\n",
    "# Step 5: Calculate Cumulative Runs, Outs, and Batting Average\n",
    "# Calculate cumulative totals and batting average for each player over the seasons.\n",
    "cumulative_runs_outs_df = seasonal_runs_outs_df \\\n",
    "    .withColumn(\"cumulative_runs\", F.sum(\"total_runs\").over(player_season_window_spec)) \\\n",
    "    .withColumn(\"cumulative_outs\", F.sum(\"total_outs\").over(player_season_window_spec)) \\\n",
    "    .withColumn(\"cumulative_avg\", F.round(F.col(\"cumulative_runs\") / F.col(\"cumulative_outs\"), 2))\n",
    "\n",
    "# Step 6: Join with Player Data to Include Detailed Player Information\n",
    "# Join the cumulative data with player information, including player name and country.\n",
    "player_cumulative_avg_df = cumulative_runs_outs_df \\\n",
    "    .join(player_df, player_df[\"player_id\"] == cumulative_runs_outs_df[\"striker\"], \"inner\") \\\n",
    "    .select(\"player_name\", \"country_name\", \"season\", \"cumulative_avg\")\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/top_batters_with_best_cumulative_avg\"\n",
    "player_cumulative_avg_df.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6da3d8aa-1a14-4db2-b7e1-64f57c0a7754",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Analyze team performance when batting first versus chasing\n",
    "\n",
    "# Step 1: Select Relevant Columns for Analysis\n",
    "# Create a DataFrame with essential columns for determining match outcomes based on toss and innings choices.\n",
    "winning_teams_df = match_df \\\n",
    "    .select(\"match_id\", \"season_year\", \"toss_winner\", \"match_winner\", \"toss_name\")\n",
    "\n",
    "# Step 2: Filter and Aggregate Data for Teams Winning When Chasing\n",
    "# Filter for matches where the winning team chased. If the toss winner chose to bat but did not win, or if the toss winner\n",
    "# chose to field and won, then the team was chasing. Group by season and match winner to count wins while chasing.\n",
    "chasing_first_winning_teams_df = winning_teams_df \\\n",
    "    .filter(\n",
    "        ((F.col(\"toss_winner\") != F.col(\"match_winner\")) & (F.col(\"toss_name\") == 'bat')) |  # Toss winner batted, but lost\n",
    "        ((F.col(\"toss_winner\") == F.col(\"match_winner\")) & (F.col(\"toss_name\") == 'field'))  # Toss winner fielded and won\n",
    "    ) \\\n",
    "    .groupBy(\"season_year\", \"match_winner\") \\\n",
    "    .agg(F.count(\"*\").alias(\"chasing_win_count\")) \\\n",
    "    .orderBy(F.col(\"season_year\").asc(), F.col(\"chasing_win_count\").desc()) \n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/winning_teams_by_chasing\"\n",
    "chasing_first_winning_teams_df.write.csv(output_path, header=True, mode=\"overwrite\")\n",
    "\n",
    "# Step 3: Filter and Aggregate Data for Teams Winning When Batting First\n",
    "# Filter for matches where the winning team batted first. If the toss winner chose to field but did not win, or if the toss \n",
    "# winner chose to bat and won, then the team batted first. Group by season and match winner to count wins while batting first.\n",
    "batting_first_winning_teams_df = winning_teams_df \\\n",
    "    .filter(\n",
    "        ((F.col(\"toss_winner\") != F.col(\"match_winner\")) & (F.col(\"toss_name\") == 'field')) |  # Toss winner fielded, but lost\n",
    "        ((F.col(\"toss_winner\") == F.col(\"match_winner\")) & (F.col(\"toss_name\") == 'bat'))      # Toss winner batted and won\n",
    "    ) \\\n",
    "    .groupBy(\"season_year\", \"match_winner\") \\\n",
    "    .agg(F.count(\"*\").alias(\"batting_win_count\")) \\\n",
    "    .orderBy(F.col(\"season_year\").asc(), F.col(\"batting_win_count\").desc()) \n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/winning_teams_by_batting_first\"\n",
    "batting_first_winning_teams_df.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a8479d33-483a-4d6b-823b-f05839ae44c0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Analyze top bowlers based on the number of wickets taken each season\n",
    "\n",
    "# Filter for deliveries resulting in a wicket, group by season, team, and bowler, then count total wickets.\n",
    "top_bowlers_by_wickets_df = ball_by_ball_df \\\n",
    "    .filter(F.col(\"bowler_wicket\") == 'true') \\\n",
    "    .groupBy(\"season\", \"team_bowling\", \"bowler\") \\\n",
    "    .agg(F.count(\"*\").alias(\"wicket_taken_season_wise\")) \\\n",
    "    .join(player_df, ball_by_ball_df[\"bowler\"] == player_df[\"player_id\"], \"inner\") \\\n",
    "    .join(team_df, ball_by_ball_df[\"team_bowling\"] == team_df[\"team_id\"], \"inner\") \\\n",
    "    .select(\"season\", \"player_name\", \"team_name\", \"country_name\", \"wicket_taken_season_wise\") \n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/top_bowlers_by_wickets\"\n",
    "top_bowlers_by_wickets_df.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f7fdfcd6-f07b-4a7f-9fa7-09b3c948589b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Analyze top bowlers based on their economy rate\n",
    "\n",
    "# Step 1: Calculate Runs per Ball for Each Bowler\n",
    "# Select relevant columns and calculate runs scored per ball, including extras.\n",
    "bowlers_economy_df = ball_by_ball_df \\\n",
    "    .select(\"season\", \"bowler\", \"team_bowling\", \"runs_scored\", \"bowler_extras\") \\\n",
    "    .withColumn('runs_per_ball', F.col(\"runs_scored\") + F.col(\"bowler_extras\"))\n",
    "\n",
    "# Step 2: Calculate Overall Economy Rate for Each Bowler\n",
    "# Group by team and bowler, sum the runs per ball and count the number of balls bowled.\n",
    "top_bowlers_economy_all_time_df = bowlers_economy_df \\\n",
    "    .groupBy(\"team_bowling\" ,\"bowler\") \\\n",
    "    .agg(F.sum(\"runs_per_ball\").alias(\"sum_runs_per_ball\"), F.count(\"*\").alias(\"count_balls\")) \\\n",
    "    .filter(F.col(\"count_balls\") >= 20) \\\n",
    "    .withColumn('overs_bowled', F.round(( F.col(\"count_balls\") / 6 ), 2)) \\\n",
    "    .withColumn('bowler_economy', F.round(( F.col(\"sum_runs_per_ball\") / F.col(\"overs_bowled\") ), 2)) \\\n",
    "    .join(player_df, ball_by_ball_df[\"bowler\"] == player_df[\"player_id\"], \"inner\") \\\n",
    "    .join(team_df, ball_by_ball_df[\"team_bowling\"] == team_df[\"team_id\"], \"inner\") \\\n",
    "    .select(\"team_name\", \"player_name\", \"country_name\", \"bowler_economy\") \\\n",
    "    .orderBy(F.col(\"bowler_economy\").asc())\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/top_bowlers_economy_all_time\"\n",
    "top_bowlers_economy_all_time_df.write.csv(output_path, header=True, mode=\"overwrite\")\n",
    "\n",
    "# Step 3: Calculate Economy Rate per Bowler for Each Season\n",
    "top_bowlers_economy_per_season_df = bowlers_economy_df \\\n",
    "    .groupBy(\"season\", \"team_bowling\" ,\"bowler\") \\\n",
    "    .agg(F.sum(\"runs_per_ball\").alias(\"sum_runs_per_ball\"), F.count(\"*\").alias(\"count_balls\")) \\\n",
    "    .filter(F.col(\"count_balls\") >= 20) \\\n",
    "    .withColumn('overs_bowled', F.round(( F.col(\"count_balls\") / 6 ), 2)) \\\n",
    "    .withColumn('bowler_economy', F.round(( F.col(\"sum_runs_per_ball\") / F.col(\"overs_bowled\") ), 2)) \\\n",
    "    .join(player_df, ball_by_ball_df[\"bowler\"] == player_df[\"player_id\"], \"inner\") \\\n",
    "    .join(team_df, ball_by_ball_df[\"team_bowling\"] == team_df[\"team_id\"], \"inner\") \\\n",
    "    .orderBy(F.col(\"season\"), F.col(\"bowler_economy\").asc()) \\\n",
    "    .select(\"season\", \"team_name\", \"player_name\", \"country_name\", \"bowler_economy\")\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/top_bowlers_economy_per_season\"\n",
    "top_bowlers_economy_per_season_df.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a8ada045-28a3-432c-91ec-46d498697357",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Analyze the most common types of dismissals in cricket\n",
    "\n",
    "# Filter for deliveries that resulted in a wicket, group by season and out type, then count occurrences.\n",
    "common_out_types_df = ball_by_ball_df \\\n",
    "    .filter(F.col(\"bowler_wicket\") == 'true') \\\n",
    "    .groupBy(\"season\", \"out_type\") \\\n",
    "    .agg(F.count(\"*\").alias(\"wicket_taken_season_out_type_wise\")) \\\n",
    "    .select(\"season\", \"out_type\", \"wicket_taken_season_out_type_wise\")\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/common_out_types\"\n",
    "common_out_types_df.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "93ecc04b-8b77-48ab-8548-7158b82fdbb8",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Analyze which bowlers excel at specific dismissal types\n",
    "\n",
    "# Filter for deliveries that resulted in a wicket, group by season, team, bowler, and out type, then count occurrences.\n",
    "bowler_wickets_out_type_wise_df = ball_by_ball_df \\\n",
    "    .filter(F.col(\"bowler_wicket\") == 'true') \\\n",
    "    .groupBy(\"season\", \"team_bowling\" ,\"bowler\", \"out_type\") \\\n",
    "    .agg(F.count(\"*\").alias(\"wicket_taken_season_out_type_wise\")) \\\n",
    "    .join(player_df, ball_by_ball_df[\"bowler\"] == player_df[\"player_id\"], \"inner\") \\\n",
    "    .join(team_df, ball_by_ball_df[\"team_bowling\"] == team_df[\"team_id\"], \"inner\") \\\n",
    "    .select(\"season\", \"player_name\", \"team_name\", \"country_name\", \"out_type\", \"wicket_taken_season_out_type_wise\")\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/bowler_wickets_out_type_wise\"\n",
    "bowler_wickets_out_type_wise_df.write.csv(output_path, header=True, mode=\"overwrite\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b2f34280-a119-443b-bf7d-c4c0dfcd4de8",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Analyze how a team's toss decision (bat or field) impacts match outcomes\n",
    "\n",
    "# Step 1: Select Relevant Columns and Filter for Winning Teams Based on Toss Decision\n",
    "toss_outcome_df = match_df \\\n",
    "    .select(\"season_year\", \"toss_winner\", \"match_winner\", \"toss_name\") \\\n",
    "    .filter(F.col(\"toss_winner\") == F.col(\"match_winner\")) \\\n",
    "    .withColumnRenamed(\"season_year\",\"season\") \\\n",
    "    .withColumnRenamed(\"toss_winner\",\"team_name\") \\\n",
    "\n",
    "# Step 2: Calculate Wins for Teams that Elected to Field\n",
    "winning_teams_by_fielding_df = toss_outcome_df \\\n",
    "    .filter(F.col(\"toss_name\") == \"field\") \\\n",
    "    .groupBy(\"season\", \"team_name\") \\\n",
    "    .agg(F.count(\"*\").alias(\"match_wins_by_fielding\")) \\\n",
    "    .withColumnRenamed(\"season\", \"field_season\") \\\n",
    "    .withColumnRenamed(\"team_name\", \"field_team_name\")\n",
    "\n",
    "# Step 3: Calculate Wins for Teams that Elected to Bat\n",
    "winning_teams_by_batting_df = toss_outcome_df \\\n",
    "    .filter(F.col(\"toss_name\") == \"bat\") \\\n",
    "    .groupBy(\"season\", \"team_name\") \\\n",
    "    .agg(F.count(\"*\").alias(\"match_wins_by_batting\")) \\\n",
    "    .withColumnRenamed(\"season\", \"bat_season\") \\\n",
    "    .withColumnRenamed(\"team_name\", \"bat_team_name\")\n",
    "\n",
    "# Step 4: Join Fielding and Batting Win Dataframes to Analyze Toss Impact\n",
    "teams_toss_effect_on_type = winning_teams_by_fielding_df \\\n",
    "    .join(\n",
    "        winning_teams_by_batting_df, \n",
    "        (winning_teams_by_fielding_df[\"field_season\"] == winning_teams_by_batting_df[\"bat_season\"]) & \n",
    "        (winning_teams_by_fielding_df[\"field_team_name\"] == winning_teams_by_batting_df[\"bat_team_name\"]), \n",
    "        \"inner\"\n",
    "    ) \\\n",
    "    .withColumn(\"total_wins\", F.col(\"match_wins_by_fielding\") + F.col(\"match_wins_by_batting\")) \\\n",
    "    .withColumn(\"win_percentage_field\", F.round(F.col(\"match_wins_by_fielding\") / F.col(\"total_wins\") * 100)) \\\n",
    "    .withColumn(\"win_percentage_bat\", 100 - F.col(\"win_percentage_field\")) \\\n",
    "    .select(\n",
    "        winning_teams_by_batting_df[\"bat_season\"].alias(\"season\"), \n",
    "        winning_teams_by_batting_df[\"bat_team_name\"].alias(\"team_name\"), \n",
    "        \"win_percentage_field\",\n",
    "        \"win_percentage_bat\"\n",
    "    ) \\\n",
    "    .orderBy(F.col(\"season\").desc())\n",
    "\n",
    "# Step 5: Analyze Teams Winning by Fielding\n",
    "teams_winning_by_fielding = teams_toss_effect_on_type \\\n",
    "    .select(\n",
    "        \"season\", \n",
    "        \"team_name\", \n",
    "        \"win_percentage_field\"\n",
    "    ) \\\n",
    "    .orderBy(F.col(\"win_percentage_field\").desc())\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/teams_winning_by_fielding\"\n",
    "teams_winning_by_fielding.write.csv(output_path, header=True, mode=\"overwrite\")\n",
    "\n",
    "\n",
    "# Step 6: Analyze Teams Winning by Batting\n",
    "teams_winning_by_batting = teams_toss_effect_on_type \\\n",
    "    .select(\n",
    "        \"season\", \n",
    "        \"team_name\", \n",
    "        \"win_percentage_bat\"\n",
    "    ) \\\n",
    "    .orderBy(F.col(\"win_percentage_bat\").desc())\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/teams_winning_by_batting\"\n",
    "teams_winning_by_batting.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "df10f96d-daba-4c42-ac94-83266d8fad1d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Analyze which teams consistently perform well across seasons\n",
    "\n",
    "# Step 1: Count occurrences where the team is team1\n",
    "team1_count_df = match_df.groupBy(\"season_year\", \"team1\") \\\n",
    "    .agg(F.count(\"*\").alias(\"team1_count\")) \\\n",
    "    .withColumnRenamed(\"team1\", \"team_name\")\n",
    "\n",
    "# Step 2: Count occurrences where the team is team2\n",
    "team2_count_df = match_df.groupBy(\"season_year\", \"team2\") \\\n",
    "    .agg(F.count(\"*\").alias(\"team2_count\")) \\\n",
    "    .withColumnRenamed(\"team2\", \"team_name\")\n",
    "\n",
    "# Step 3: Count winning occurrences for each team\n",
    "winning_team_count = match_df.groupBy(\"season_year\", \"match_winner\") \\\n",
    "    .agg(F.count(\"*\").alias(\"match_win_count\")) \\\n",
    "    .withColumnRenamed(\"match_winner\", \"team_name\")\n",
    "\n",
    "# Step 4: Merge counts to get total matches played by each team in each season\n",
    "total_matches_df = team1_count_df.join(\n",
    "    team2_count_df,\n",
    "    on=[\"season_year\", \"team_name\"],\n",
    "    how=\"outer\"\n",
    ").join(\n",
    "    winning_team_count, \n",
    "    on=[\"season_year\", \"team_name\"],\n",
    "    how=\"outer\"\n",
    ") \\\n",
    ".fillna(0, [\"team1_count\", \"team2_count\"]) \\\n",
    ".withColumn(\"total_matches\", F.col(\"team1_count\") + F.col(\"team2_count\")) \\\n",
    ".withColumn(\"winning_percentage\", F.round((F.col(\"match_win_count\")/F.col(\"total_matches\")*100), 2) ) \\\n",
    ".select(\"season_year\", \"team_name\", \"total_matches\", \"match_win_count\", \"winning_percentage\")\n",
    "\n",
    "\n",
    "# Step 5: Save the results\n",
    "total_matches_df.orderBy(\"season_year\", F.col(\"winning_percentage\").desc())\n",
    "\n",
    "output_path = \"dbfs:/FileStore/tables/teams_winning_percentage\"\n",
    "total_matches_df.write.csv(output_path, header=True, mode=\"overwrite\")"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "environmentMetadata": null,
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "IPL_Data_Preparation",
   "widgets": {}
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}