soccer_player.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* UDF\n",
    "* Join\n",
    "* Windowing\n",
    "- How a spark script is submitted"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('soccer_player').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "player_file = '../datasets/player.csv'\n",
    "player_attr_file = '../datasets/player_attributes.csv'\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: string (nullable = true)\n",
      " |-- player_api_id: string (nullable = true)\n",
      " |-- player_name: string (nullable = true)\n",
      " |-- player_fifa_api_id: string (nullable = true)\n",
      " |-- birthday: string (nullable = true)\n",
      " |-- height: string (nullable = true)\n",
      " |-- weight: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "player_df = spark.read.format('csv').option('header','true').load(player_file)\n",
    "\n",
    "player_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "pa_df = spark.read.csv(player_attr_file, header=True)\n",
    "pa_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "player_df.count(), pa_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "player_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## UDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import udf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pa_df.select('date').show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['acceleration',\n",
       " 'aggression',\n",
       " 'agility',\n",
       " 'attacking_work_rate',\n",
       " 'balance',\n",
       " 'ball_control',\n",
       " 'crossing',\n",
       " 'curve',\n",
       " 'defensive_work_rate',\n",
       " 'dribbling',\n",
       " 'finishing',\n",
       " 'free_kick_accuracy',\n",
       " 'gk_diving',\n",
       " 'gk_handling',\n",
       " 'gk_kicking',\n",
       " 'gk_positioning',\n",
       " 'gk_reflexes',\n",
       " 'heading_accuracy',\n",
       " 'id',\n",
       " 'interceptions',\n",
       " 'jumping',\n",
       " 'long_passing',\n",
       " 'long_shots',\n",
       " 'marking',\n",
       " 'overall_rating',\n",
       " 'penalties',\n",
       " 'player_api_id',\n",
       " 'player_fifa_api_id',\n",
       " 'positioning',\n",
       " 'potential',\n",
       " 'preferred_foot',\n",
       " 'reactions',\n",
       " 'short_passing',\n",
       " 'shot_power',\n",
       " 'sliding_tackle',\n",
       " 'sprint_speed',\n",
       " 'stamina',\n",
       " 'standing_tackle',\n",
       " 'strength',\n",
       " 'vision',\n",
       " 'volleys',\n",
       " 'year']"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "udf_year_extract = udf(lambda c: c.split('-')[0])\n",
    "#pa_df = pa_df.withColumn('year', udf_year_extract(pa_df.date)).drop('date')\n",
    "sorted(pa_df.columns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Find best striker of 2016"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "pa_2016 = pa_df.where(pa_df.year == 2016)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "pa_strikers = pa_2016.groupBy('player_api_id').agg(\n",
    "                                                    {'shot_power':'avg',\n",
    "                                                    'finishing':'avg',\n",
    "                                                    'acceleration':'avg'\n",
    "                                                    })\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-----------------+-----------------+---------------+\n",
      "|player_api_id|   avg(finishing)|avg(acceleration)|avg(shot_power)|\n",
      "+-------------+-----------------+-----------------+---------------+\n",
      "|       309726|75.44444444444444|74.11111111111111|           76.0|\n",
      "|        26112|             53.0|             51.0|           76.0|\n",
      "|        38433|            68.25|             74.0|           74.0|\n",
      "|       295060|             25.0|             62.0|           40.0|\n",
      "|       161396|             29.0|             72.0|           69.0|\n",
      "+-------------+-----------------+-----------------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pa_strikers.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Join with player information in player_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "joined_striker = player_df.select('player_api_id', 'player_name')\\\n",
    "                .join(pa_strikers, on=player_df.player_api_id == pa_strikers.player_api_id)\n",
    "joined_striker.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Another way. Avoids repetition of key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------+-----------------+-----------------+---------------+\n",
      "|player_api_id|   player_name|   avg(finishing)|avg(acceleration)|avg(shot_power)|\n",
      "+-------------+--------------+-----------------+-----------------+---------------+\n",
      "|       309726|Andrea Belotti|75.44444444444444|74.11111111111111|           76.0|\n",
      "|        26112|Benoit Cheyrou|             53.0|             51.0|           76.0|\n",
      "|        38433|  Borja Valero|            68.25|             74.0|           74.0|\n",
      "|       295060|  Daniel Potts|             25.0|             62.0|           40.0|\n",
      "|       161396|Diego Contento|             29.0|             72.0|           69.0|\n",
      "+-------------+--------------+-----------------+-----------------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joined_striker = player_df.select('player_api_id', 'player_name')\\\n",
    "                .join(pa_strikers, on=['player_api_id'])\n",
    "joined_striker.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Broadcast join\n",
    "Broadcast smaller dataframe to make a copy on every executor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-----------------+\n",
      "|player_api_id|      total_score|\n",
      "+-------------+-----------------+\n",
      "|       309726|75.38888888888889|\n",
      "|        26112|             64.0|\n",
      "|        38433|          72.5625|\n",
      "|       295060|            41.75|\n",
      "|       161396|            59.75|\n",
      "+-------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "udf_overall_score = udf(lambda f,a,s : (f+a+2*s)/4)\n",
    "pa_strikers_2016 = pa_strikers.withColumn('total_score', udf_overall_score('avg(finishing)','avg(acceleration)','avg(shot_power)'))\\\n",
    "                .drop('avg(finishing)','avg(acceleration)','avg(shot_power)')\n",
    "\n",
    "pa_strikers_2016.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(5586, 11060)"
      ]
     },
     "execution_count": 86,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pa_strikers_2016.count(), player_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+------------------+-----------------+\n",
      "|player_api_id|       player_name|      total_score|\n",
      "+-------------+------------------+-----------------+\n",
      "|       505942|Aaron Appindangoye|             53.5|\n",
      "|       155782|   Aaron Cresswell|68.41666666666666|\n",
      "|       162549|       Aaron Doran|            64.25|\n",
      "|        30572|     Aaron Galindo|             43.0|\n",
      "|        27316|        Aaron Hunt|            74.75|\n",
      "+-------------+------------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import broadcast\n",
    "joined_striker = player_df.select(\n",
    "                                'player_api_id', \n",
    "                                'player_name')\\\n",
    "                        .join(\n",
    "                                broadcast(pa_strikers_2016),\n",
    "                                on=['player_api_id'],\n",
    "                                how='inner')\n",
    "joined_striker.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------------+-----------------+\n",
      "|player_api_id|         player_name|      total_score|\n",
      "+-------------+--------------------+-----------------+\n",
      "|        20276|                Hulk|            89.25|\n",
      "|        37412|       Sergio Aguero|             89.0|\n",
      "|        38817|        Carlos Tevez|            88.75|\n",
      "|        32118|      Lukas Podolski|            88.25|\n",
      "|        31921|         Gareth Bale|             87.0|\n",
      "|        30834|        Arjen Robben|            86.75|\n",
      "|       303824|       Memphis Depay|85.10714285714286|\n",
      "|       129944|          Marco Reus|             85.0|\n",
      "|       158263|        Dorlan Pabon|            84.75|\n",
      "|       150565|Pierre-Emerick Au...|            84.75|\n",
      "|        25759|     Gonzalo Higuain|84.66666666666667|\n",
      "|       156726|       Douglas Costa|             84.5|\n",
      "|       169193| Alexandre Lacazette|          84.4375|\n",
      "|       286119|         Jamie Vardy|84.42857142857143|\n",
      "|        30348|       Jermain Defoe|           84.375|\n",
      "|        46509|      Angel Di Maria|            84.25|\n",
      "|        93447|  Robert Lewandowski|            84.25|\n",
      "|        50047|      Alexis Sanchez|            84.25|\n",
      "|       178812|     Andre Schuerrle|             84.0|\n",
      "|       181276|       Romelu Lukaku|             84.0|\n",
      "+-------------+--------------------+-----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joined_striker.sort(joined_striker.total_score.desc()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Find if height is related to heading accuracy of players"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 1. join heading accuracy with player height and create a dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(183978, 11060)"
      ]
     },
     "execution_count": 91,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pa_df.count(),player_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [],
   "source": [
    "player_heading_acc = pa_df.select('player_api_id','heading_accuracy')\\\n",
    "                        .join(\n",
    "                            broadcast(player_df.select('player_api_id','height')), \n",
    "                            on=['player_api_id']\n",
    "                            )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----------------+------+\n",
      "|player_api_id|heading_accuracy|height|\n",
      "+-------------+----------------+------+\n",
      "|       505942|              71|182.88|\n",
      "|       505942|              71|182.88|\n",
      "|       505942|              71|182.88|\n",
      "|       505942|              70|182.88|\n",
      "|       505942|              70|182.88|\n",
      "+-------------+----------------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "player_heading_acc = player_heading_acc.dropna()\n",
    "player_heading_acc.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Accumulators"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Height"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "metadata": {},
   "outputs": [],
   "source": [
    "short_count = spark.sparkContext.accumulator(0)\n",
    "medium_low_count = spark.sparkContext.accumulator(0)\n",
    "medium_high_count = spark.sparkContext.accumulator(0)\n",
    "high_count = spark.sparkContext.accumulator(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {},
   "outputs": [],
   "source": [
    "def count_players_by_height(r):\n",
    "    height = float(r.height)\n",
    "    if height<175:\n",
    "        short_count.add(1)\n",
    "    elif height<183:\n",
    "        medium_low_count.add(1)\n",
    "    elif height<195:\n",
    "        medium_high_count.add(1)\n",
    "    else:\n",
    "        high_count.add(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {},
   "outputs": [],
   "source": [
    "player_heading_acc.foreach(count_players_by_height)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(19128, 98541, 62081, 3392)"
      ]
     },
     "execution_count": 132,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "short_count.value, medium_low_count.value, medium_high_count.value, high_count.value"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {},
   "outputs": [],
   "source": [
    "short_ha_count = spark.sparkContext.accumulator(0)\n",
    "medium_low_ha_count = spark.sparkContext.accumulator(0)\n",
    "medium_high_ha_count = spark.sparkContext.accumulator(0)\n",
    "high_ha_count = spark.sparkContext.accumulator(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {},
   "outputs": [],
   "source": [
    "def count_player_acc_count_by_height(r, threshold):\n",
    "    height = float(r.height)\n",
    "    acc= float(r.heading_accuracy)\n",
    "    \n",
    "    if acc< threshold:\n",
    "        return\n",
    "    if height<175:\n",
    "        short_ha_count.add(1)\n",
    "    elif height<183:\n",
    "        medium_low_ha_count.add(1)\n",
    "    elif height<195:\n",
    "        medium_high_ha_count.add(1)\n",
    "    else:\n",
    "        high_ha_count.add(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 135,
   "metadata": {},
   "outputs": [],
   "source": [
    "# calculate player count with >60% accuracy\n",
    "player_heading_acc.foreach(lambda x: count_player_acc_count_by_height(x,60))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(4290, 46086, 41785, 1580)"
      ]
     },
     "execution_count": 137,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "short_ha_count.value,medium_low_ha_count.value,medium_high_ha_count.value,high_ha_count.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[22.42785445420326, 46.76835022985356, 67.30722765419371, 46.58018867924528]"
      ]
     },
     "execution_count": 138,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "[short_ha_count.value/short_count.value*100,\n",
    " medium_low_ha_count.value/medium_low_count.value*100,\n",
    " medium_high_ha_count.value/medium_high_count.value*100,\n",
    " high_ha_count.value/high_count.value*100]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}