pyspark_workshop.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "<div style=\"float: right;margin-right: 1em; margin-bottom: 3em;\">\n",
    "<a href=\"https://pybcn.org\">\n",
    "   <img src=\"Images/pybcn.png\" width=50>\n",
    "</a>\n",
    "    \n",
    "<a href=\"https://engineering.affectv.com\">\n",
    "    <img src=\"Images/affectv.png\" width=100 style=\"margin-top: 1em;\">\n",
    "</a>\n",
    "</div>\n",
    "\n",
    "# The Magic of **Py**Spark\n",
    "\n",
    "### _An introduction to PySpark_\n",
    "\n",
    "&nbsp;\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You only need to do the following if you are running in Binder (or have not downloaded the soccer dataset). This is around 5MB"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import urllib.request\n",
    "url = \"https://projects.fivethirtyeight.com/soccer-api/club/spi_matches.csv\"\n",
    "data = urllib.request.urlopen(url).read()\n",
    "\n",
    "with open(\"spi_matches.csv\", \"wb\") as f:\n",
    "    f.write(data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating a `SparkSession` and loading some data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first step is to create a `SparkSession` object we can interact with. It is customary to call it `spark`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since the dataset is in the local filesystem, **we need to load it**. Not only Spark comes with many different types of loaders, but also handles parallelisation: if our dataset was stored in a distributed store (like HDFS or S3), each worker would fetch only a piece of the data.\n",
    "\n",
    "You can pass additional `option` to readers, in this case I know this CSV file comes with headers, hence I want the reader to handle the first row differently. We could also pass a different delimiter (like tabs, or pipes) or what to do with quotes. Check the documentation of each reader to know what options are available, there are many!\n",
    "\n",
    "*NOTE*: when loading distributed CSVs, each CSV file should come with the header. Also note that CSV is a horrible format to use!\n",
    "\n",
    "*TIP*: [other loaders](https://spark.apache.org/docs/latest/sql-data-sources.html) (or, datasources as they are known in Spark) have pretty straightforward names, like [`.parquet()`](https://spark.apache.org/docs/latest/sql-data-sources-parquet.html) and [`.json()`](https://spark.apache.org/docs/latest/sql-data-sources-json.html) (the latter expects one-json-per-line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe = spark.read.option(\"header\", True).csv(\"spi_matches.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We don't know much about this file aside from it being CSV. First we can do is to check what the columns are named. The advantage of reading with the read method is that all will come tagged with names, a disadvantage may be the types: everything will show as `string`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we may want to see how the data looks like. In general you would use the [`show`](https://spark.apache.org/docs/2.4.4/api/python/pyspark.sql.html#pyspark.sql.DataFrame.show) method on a dataframe, which is similar to Pandas'. We can also use [`describe`](https://spark.apache.org/docs/2.4.4/api/python/pyspark.sql.html#pyspark.sql.DataFrame.describe) to compute some statistics per column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "dataframe.describe().show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You may have noticed that the second operation (`describe().show()`) was slower. Why was that? Spark operations are split in _actions_ and _transformations_. \n",
    "- *Transformations*: \"explain\" or \"define\" an operation. When you do `desc = df.describe()`, you are just stating _desc should contain the describe operation on df_. That's just a description, and takes essentially zero time: you are not requesting to _see it_ or _store it_ or...\n",
    "- *Actions*: Actions \"require\" something to be done. They show, store, or in some sense, operate (in functional programming you'd call them _effectful_)\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And, yes, the output looks a bit horrible `¯\\_(ツ)_/¯`. It is clear though that some of these columns are actually float/double, others are ints and others are indeed strings. And the first looks pretty much a datetime. First step would be to change the types to the correct ones. \n",
    "\n",
    "There are several ways to do this, the easiest one is to just select what we want and cast"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Converting column datatypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe_with_1_type = dataframe.select(dataframe[\"league_id\"].cast(\"int\"))\n",
    "dataframe_with_1_other_type = dataframe.select(dataframe[\"xg1\"].cast(\"float\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In Spark, there are several ways of selecting a column:\n",
    "- by name\n",
    "like `select(\"columnName\")` or `select(col(\"columnName\"))`\n",
    "- by \"named reference\"\n",
    "like `select(dataframe.columnName)` or `select(dataframe[\"columnName\"]`\n",
    "- _if you were using Scala you'd have two more_\n",
    "\n",
    "The first is more generic: it is referencing _a_ column with that name, but _not necessarily in a specific dataframe_. This distinction can be important in some machine learning algorithms, where additional metadata is added _to a column IN a dataframe_. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe_with_1_type.printSchema()\n",
    "dataframe_with_1_type.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe_with_1_other_type.printSchema()\n",
    "dataframe_with_1_other_type.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "What would happen if we messed up with the types…?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe_with_wrong_type = dataframe.select(dataframe[\"xg1\"].cast(\"int\"))\n",
    "dataframe_with_wrong_type.printSchema()\n",
    "dataframe_with_wrong_type.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In general, for types offering it, casting will either truncate or convert. So, going from _should be a float_ to `int` will just round. This won't be the case for types not offering it though, the conversion will be _really bad_."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe_with_very_wrong_type = dataframe.select(dataframe[\"league\"].cast(\"int\"))\n",
    "dataframe_with_very_wrong_type.printSchema()\n",
    "dataframe_with_very_wrong_type.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that using strings is error prone: what if you typed `cast(\"fioat\")`? It is better to use the built in type objects"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import FloatType, IntegerType, StringType, DateType\n",
    "\n",
    "dataframe_with_1_type_2 = dataframe.select(dataframe[\"league_id\"].cast(IntegerType()))\n",
    "dataframe_with_1_type_2.printSchema()\n",
    "dataframe_with_1_type_2.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we don't really need all columns from this dataframe, but only the following:\n",
    "\n",
    "- date\n",
    "- league\n",
    "- team1\n",
    "- team2 \n",
    "- score1\n",
    "- score2\n",
    "\n",
    "We can select on a list, as easily as `select(col_1, col_2)`, and we'd also like to convert to the correct types on the go. Shouldn't be hard in Python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conv = {\n",
    "    \"date\": DateType(),\n",
    "    \"league\": StringType(),\n",
    "    \"team1\": StringType(),\n",
    "    \"team2\": StringType(),\n",
    "    \"score1\": IntegerType(),\n",
    "    \"score2\": IntegerType(),\n",
    "}\n",
    "\n",
    "selection = [dataframe[k].cast(v) for k, v in conv.items()]\n",
    "\n",
    "df = dataframe.select(*selection)  # reminder: *list unpacks a list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df.printSchema()\n",
    "df.show(5)\n",
    "df.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Note*: we could have used the `inferSchema` setting in the CSV loader, though. But:\n",
    "- We wouldn't have an excuse to explain how to manually type columns\n",
    "- It makes _two_ passes on the data, which is usually one more than needed\n",
    "- It may still mess up if there is some string lurking"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe_inferred = spark.read.option(\"header\", True).option(\"inferSchema\", True).csv(\"spi_matches.csv\")\n",
    "dataframe_inferred.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we may want to filter a specific league, like the Premier League, and count how many matches are there"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Operating on data (filtering, transforming, ordering...), SparkSQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You should eventually be familiar, a bit, with what is available in `pyspark.sql.functions`. Even when working in Python, [I check the Scala reference](https://spark.apache.org/docs/2.4.4/api/scala/index.html#org.apache.spark.sql.functions$) instead of the Python one (maybe because I usually work in Scala, though)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col\n",
    "\n",
    "premier = df.where(col(\"league\").contains(\"Barclays\"))\n",
    "premier.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That's a lot of matches, I'd like to figure out how many years of data we have"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import date_trunc\n",
    "\n",
    "premier.select(date_trunc(\"year\", col(\"date\")).alias(\"year\")).distinct().orderBy(col(\"date\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "I have introduced many things here! For one, a function to work on a column, using distinct and ordering. But they should be pretty clear, this looks mostly like SQL, but as functions. Huh, wait... this reminds me..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "premier.createOrReplaceTempView(\"premier\")\n",
    "\n",
    "spark.sql(\"SELECT DISTINCT date_trunc('year', date) AS year FROM premier ORDER BY year\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see, Spark offers **SparkSQL**, allowing you to write your queries in pretty much standard SQL (there are some minor differences) instead of as methods on a dataframe. First you register your tables, then you operate on them\n",
    "\n",
    "*NOTE*: Using methods on a dataframe is a better approach in general, since modularity, testing and specially _parsing_ are important. See next cell"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# try to run this\n",
    "\n",
    "spark.sql(\"SELCT DISTINCT date_trunc('year', date) AS year FROM premier ORDER BY year\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It will fail very badly (with a stacktrace from Scala, showing that you referenced a non-existing SQL method). This error should be caught by pylint/mypy/pyflakes or any other Python syntax checker, if you use the methods on a dataframe and write `df.selct` instead of `df.select`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we want to do anything with the data, we don't want matches in the future. Let's check if we can filter them out somehow without needing to know when we downloaded this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "premier.where(\"date_trunc('year', date) = '2020'\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see, you can use SQL-like expressions in `where` method calls, too. Like above, this should be discouraged, but when exploring data it is faster to type, so it's not that bad. But make sure not to use it in production code!\n",
    "\n",
    "Also, _clearly_ we can remove future matches by searching for null scores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "current_premier = premier.where(\"score1 is not null\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "current_premier.select(date_trunc(\"year\", col(\"date\")).alias(\"year\")).distinct().orderBy(col(\"date\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "_Clearly_ is, in general, not good enough. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "premier.where(\"score1 is null\")\\\n",
    "       .select(date_trunc(\"year\", col(\"date\")).alias(\"year\")).distinct().orderBy(col(\"date\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Did we find some weird anomaly in the data? How does it look like? Here we use again SQL-where, which is faster to write when exploring data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "premier.where(\"date_trunc('year', date) = '2019' and score1 is null\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Oh, of course: `date_trunc` just keeps the year and sets 1st of January. The rest of the current year should have no data, obviously. Let's confirm this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "premier.where(\"date_trunc('year', date) = '2019' and score1 is null\").groupBy(\"date\").count().orderBy(\"date\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Makes sense, actually. If this was part of a data pipeline, this last check should be part of the code, with some business-level validation. For this to make sense, you need to know your data, your data's SLAs, etc: in this case the validation could have been \"no nulls for dates before today\".\n",
    "\n",
    "To be fair, though, this has been a pain of copy pasting, and copy pasting is in general... bad. How would we do this in a function?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def years_in_data(dataframe):\n",
    "    return dataframe.select(date_trunc(\"year\", col(\"date\")).alias(\"year\")).distinct().orderBy(col(\"date\"))\n",
    "\n",
    "years_in_data(premier).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Easy peasy, dataframes are in the end just Python objects we can pass around, manipulate, etc. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Groups, Aggregates, using `explain`\n",
    "\n",
    "Let's do some basic analysis: which team has won more matches in this period?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import when\n",
    "\n",
    "either_winner = current_premier.select(\"team1\", \n",
    "                                       \"team2\", \n",
    "                                       when(col(\"score1\") > col(\"score2\"), True).otherwise(False).alias(\"win1\"),\n",
    "                                       when(col(\"score2\") > col(\"score1\"), True).otherwise(False).alias(\"win2\"),)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "either_winner.printSchema()\n",
    "either_winner.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "winners = either_winner.where(col(\"win1\") | col(\"win2\"))\\\n",
    "                       .withColumn(\"winner\", \n",
    "                                   when(col(\"win1\"), col(\"team1\")).otherwise(col(\"team2\")))\\\n",
    "                       .drop(\"team1\", \"team2\", \"win1\", \"win2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "winners.printSchema()\n",
    "winners.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This was a lot to digest. The process has been:\n",
    "- Create a boolean column for home win\n",
    "- Create a boolean column for away win\n",
    "- Filter out draws\n",
    "- Pick the winner according to who was the home or away team\n",
    "\n",
    "*Note*: This could have been done just with a `when` query, or using many other options, but I wanted to show a few more operations, also putting everything in a single `when` whould have been unreadable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import desc\n",
    "\n",
    "best = winners.groupBy(\"winner\").count().orderBy(desc(\"count\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You may be thinking\n",
    "> I have heard grouping is bad in Spark, because it is very slow.\n",
    "\n",
    "That's not wrong, but is not 100% right either. `groupByKey` is the bad one (on RDDs), because it actually builds the grouped data: that is in general very bad. `groupBy` generates only a logical (\"transformation style\") grouping, when you add an aggregation function (like we did by adding `count`, the generic one would be `.agg(agg_function, column)`) what will happen when an action appears is similar to a `reduceByKey`, in other words, the groups are never created:\n",
    "\n",
    "---\n",
    "\n",
    "#### Differences between `groupByKey` vs `groupBy` (or `reduceByKey`)\n",
    "\n",
    "Imagine we have this dataset: `(a, 1), (b, 2), (a, 12), (a, 13)` and we `groupBy` the first column. This first step would generate a dataset similar to `a: [(a, 1), (a, 12), (a, 13)], b: [(b, 2)]` and would require all data for key `a` to fit in one machine. If we then count, it will count the array length per key, `(a, 3, b, 1)`.\n",
    "\n",
    "`reduceByKey` would be an equivalent way of cheaply aggregating in an RDD setting (first you'd run , and we'd need to offer the `reduce` function, a function that \"does something\" for each element with that key. In this case, it would just be adding 1 per element.\n",
    "\n",
    "But, since you should probably be using Dataframes only, you'd want to use `groupBy`. When you run it, nothing happens (it is a transformation), but when we run `groupBy().count()` (followeded by an action, since counting groups is a transformation as well) the full group will never, ever, appear in memory: each time an element is \"checked\", the counter for the key will be incremented and we will directly get `(a, 3), (b, 1)`.\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "best.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This `show` took longer than the others, what is going on? To figure it out we can use `explain` (and some other methods)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "best.explain()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this we can find all the steps we have applied on top of the CSV: selection (that's `Project`), casting... There is some magic going here, with filters and operations being pushed down as closer to the data as possible. If we hadn't been `show`ing so often, the process would have been:\n",
    "- Read the data row-by-row (or batch by batch), ignore columns we have removed and apply the required `NotNull` filters before casting\n",
    "- Apply `when` conditions\n",
    "- Aggregate and count\n",
    "\n",
    "You can get additional information with\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "best.explain(extended=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "In Scala you can do some more fancy stuff with the plans, but usually you shouldn't anyway\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, imagine that we want to report on results (what we had in `premier`) together with the total number of wins for the team. And, that we don't want to use a `Window` function. I.e. we have an excuse to learn about `join` in Spark!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.autoBroadcastJoinThreshold\", -1) #  This will be explained later\n",
    "\n",
    "joined = premier.alias(\"premier\").join(best.alias(\"best1\"), col(\"best1.winner\") == col(\"premier.team1\"))\\\n",
    "                                 .withColumnRenamed(\"count\", \"team1_wins\")\\\n",
    "                                 .drop(col(\"best1.winner\"))\\\n",
    "                                 .join(best.alias(\"best2\"), col(\"best2.winner\") == col(\"premier.team2\"))\\\n",
    "                                 .withColumnRenamed(\"count\", \"team2_wins\")\\\n",
    "                                 .drop(col(\"best2.winner\")) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Writing joins without resorting to SQL needs some practice and some tips:\n",
    "- I recommend adding aliases to the tables, since it makes joins clearer, especially if there are columns _with the same name_ in both tables but you only join on a few\n",
    "- Please, please try to remove/rename columns with the same names _before_ the join, it will make your life easier\n",
    "- Note that the name of format `table.column` is only relevant for some methods (the ones using columns as inputs instead of strings). This can be a source of confusion (especially since `withColumnRenamed` doesn't fail if the column does not exist)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "joined.printSchema()\n",
    "joined.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "joined.explain()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "[`SortMergeJoin`](https://en.wikipedia.org/wiki/Sort-merge_join) is [one of the methods](https://github.com/apache/spark/tree/master/sql/core/src/main/scala/org/apache/spark/sql/execution/joins) for joining two tables, when they are sorted. In this case, the Spark query planner has decided sorting these tables for the join is the best way of doing it.\n",
    "\n",
    "*TRIVIA*: The \"gold standard\" for distributed joins is the (distributed) [hash join](https://en.wikipedia.org/wiki/Hash_join). Hash join is used locally also when data is too large for memory.\n",
    "\n",
    "In this case, Spark would also have been smart enough to use a *broadcast join*, since the tables are small (this is why I disabled it before, I wanted to show how to force it). Sometimes, it can be necessary to hint to Spark that one of the tables is significantly smaller, and broadcasting is cheaper. You'd do it as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import broadcast\n",
    "\n",
    "joined_2 = premier.alias(\"premier\").join(broadcast(best.alias(\"best1\")), col(\"best1.winner\") == col(\"premier.team1\"))\\\n",
    "                                 .withColumnRenamed(\"count\", \"team1_wins\")\\\n",
    "                                 .drop(\"winner\")\\\n",
    "                                 .join(broadcast(best.alias(\"best2\")), col(\"best2.winner\") == col(\"premier.team2\"))\\\n",
    "                                 .withColumnRenamed(\"count\", \"team2_wins\")\\\n",
    "                                 .drop(\"winner\") "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "joined_2.explain()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This hinted Spark to use a `BroadcastHashJoin` which is in general more efficient if tables are small: less shuffling means faster operations."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# UDFs (User-Defined functions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So far, we have operated mostly in the JVM. When using the Dataframe API, only `show` and `count` have sent anything significant to the Pythonk side of operations, all else has actually been running in the Spark core. That's actually pretty good! The serialization cost of sending data back and forth between Python and the JVM is large.\n",
    "\n",
    "UDFs, functions defined in Python that operate in Spark data can, thus, be _extremely_ slow, especially when compared with similar functions available in `spark.sql.functions`. There are some improvements in Spark 2.4 and greater, which we'll see now."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import BooleanType\n",
    "\n",
    "def home_win(home_score, away_score):\n",
    "  return home_score > away_score\n",
    "spark.udf.register(\"home_win_python\", home_win, BooleanType()) #  It's better if we can have the type!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "udf_example = current_premier.select(\"team1\", \"team2\", \"score1\", \"score2\")\\\n",
    "                             .withColumn(\"home_win\", home_win(col(\"score1\"), col(\"score2\")))\n",
    "\n",
    "udf_example.printSchema()\n",
    "udf_example.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "You can also use a decorator for defining UDFs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import udf\n",
    "\n",
    "@udf(BooleanType())\n",
    "def away_win(home_score, away_score):\n",
    "  return home_score < away_score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "udf_example_2 = current_premier.select(\"team1\", \"team2\", \"score1\", \"score2\")\\\n",
    "                               .withColumn(\"away_win\", away_win(col(\"score1\"), col(\"score2\")))\n",
    "\n",
    "udf_example_2.printSchema()\n",
    "udf_example_2.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A problem though is that these UDFs require moving data to and from the JVM, serializing and deserializing, which is slow. A faster way is using Pandas columnar UDFs, especially if you plan on doing any Pandas or Numpy computation\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pandas vectorized UDFs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "from pyspark.sql.functions import pandas_udf\n",
    "\n",
    "@pandas_udf(returnType=BooleanType())\n",
    "def tie(home_score, away_score):\n",
    "  return home_score == away_score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "udf_example_3 = current_premier.select(\"team1\", \"team2\", \"score1\", \"score2\")\\\n",
    "                               .withColumn(\"tie\", tie(col(\"score1\"), col(\"score2\")))\n",
    "udf_example_3.printSchema()\n",
    "udf_example_3.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "udf_example_3 = current_premier.select(\"team1\", \"team2\", \"score1\", \"score2\")\\\n",
    "                               .withColumn(\"tie\", tie(col(\"score1\"), col(\"score2\")))\n",
    "udf_example_3.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "udf_example_2 = current_premier.select(\"team1\", \"team2\", \"score1\", \"score2\")\\\n",
    "                               .withColumn(\"away_win\", away_win(col(\"score1\"), col(\"score2\")))\n",
    "udf_example_2.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This didn't look faster to me"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.execution.arrow.enabled\", \"true\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "udf_example_3 = current_premier.select(\"team1\", \"team2\", \"score1\", \"score2\")\\\n",
    "                               .withColumn(\"tie\", tie(col(\"score1\"), col(\"score2\")))\n",
    "udf_example_3.toPandas().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "udf_example_2 = current_premier.select(\"team1\", \"team2\", \"score1\", \"score2\")\\\n",
    "                               .withColumn(\"away_win\", away_win(col(\"score1\"), col(\"score2\")))\n",
    "udf_example_2.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is still not significantly faster, but this is due to how small this dataset is. Let's create some random data and compare the effect of the Arrow optimization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import rand\n",
    "\n",
    "df = spark.range(1 << 20).toDF(\"id\").withColumn(\"x\", rand()) \n",
    "# this is a large-ish random dataset, generated fully in the JVM. At least, largest that the match data!\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "spark.conf.set(\"spark.sql.execution.arrow.enabled\", \"false\")\n",
    "pandas_df = df.toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "On my run, this took `13.5s` wall time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "spark.conf.set(\"spark.sql.execution.arrow.enabled\", \"true\")\n",
    "pandas_df = df.toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And this took `686ms` for a _20x speedup_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'd see similar speedups with using `pandas_udf`s, if our match dataset was larger."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import randn, floor, abs\n",
    "\n",
    "many_matches = spark.range(1 << 24).toDF(\"id\")\\\n",
    "                    .withColumn(\"score1\", floor(abs(randn())*2.5))\\\n",
    "                    .withColumn(\"score2\", floor(abs(randn())*1.5)) # We might as well make it more natural\n",
    "\n",
    "many_matches.show(5) #  Although it would look nicer with a toPandas().head(), this might make your session explode\n",
    "many_matches.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import desc, col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.execution.arrow.enabled\", \"false\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "udf_example_no_pandas = many_matches.select(\"id\", \"score1\", \"score2\")\\\n",
    "                                    .withColumn(\"away_win\", away_win(col(\"score1\"), col(\"score2\")))\n",
    "udf_example_no_pandas.orderBy(desc(\"id\")).show() #  The orderBy is to prevent Spark from being too smart and optimising\n",
    "udf_example_no_pandas.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.execution.arrow.enabled\", \"true\")\n",
    "spark.conf.set(\"spark.sql.execution.arrow.maxRecordsPerBatch\", \"100000\")\n",
    "\n",
    "@pandas_udf(returnType=BooleanType())\n",
    "def away_win_pd(home_score, away_score):\n",
    "    return home_score < away_score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "udf_example_pandas = many_matches.select(\"id\", \"score1\", \"score2\")\\\n",
    "                                 .withColumn(\"away_win\", away_win_pd(col(\"score1\"), col(\"score2\")))\n",
    "udf_example_pandas.orderBy(desc(\"id\")).show()\n",
    "udf_example_pandas.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Without Arrow and Pandas, this took around 35-40 seconds, vs 13-15 using Arrow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) Project [id#142L, score1#146L, score2#149L, pythonUDF0#332 AS away_win#271]\n",
      "+- BatchEvalPython [away_win(score1#146L, score2#149L)], [id#142L, score1#146L, score2#149L, pythonUDF0#332]\n",
      "   +- *(1) Project [id#142L, score1#146L, FLOOR((abs(randn(3347746706199580393)) * 1.5)) AS score2#149L]\n",
      "      +- *(1) Project [id#142L, FLOOR((abs(randn(-864362105985009874)) * 2.5)) AS score1#146L]\n",
      "         +- *(1) Range (0, 16777216, step=1, splits=2)\n"
     ]
    }
   ],
   "source": [
    "udf_example_no_pandas.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) Project [id#142L, score1#146L, score2#149L, pythonUDF0#334 AS away_win#303]\n",
      "+- ArrowEvalPython [away_win_pd(score1#146L, score2#149L)], [id#142L, score1#146L, score2#149L, pythonUDF0#334]\n",
      "   +- *(1) Project [id#142L, score1#146L, FLOOR((abs(randn(3347746706199580393)) * 1.5)) AS score2#149L]\n",
      "      +- *(1) Project [id#142L, FLOOR((abs(randn(-864362105985009874)) * 2.5)) AS score1#146L]\n",
      "         +- *(1) Range (0, 16777216, step=1, splits=2)\n"
     ]
    }
   ],
   "source": [
    "udf_example_pandas.explain()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see, the only practical difference is `ArrowEvalPython` instead of `BatchEvalPython`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The history server\n",
    "\n",
    "If you are running locally (or in a dedicated cluster) you can access either master:4040 or master:18080 (or master:280888) to visit [Spark History Server](https://spark.apache.org/docs/latest/monitoring.html) where you can explore different tabs of information"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"Images/JobsView.png\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"Images/StagesView.png\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here you can see two queries of the SQL tab expanded (do they look familiar?)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"Images/QueryPlan1.png\" width=300 style=\"float:left;\"> <img src=\"Images/QueryPlan2.png\" width=300>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deployment and other tips"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Launching jobs\n",
    "\n",
    "For both Scala Spark and PySpark, deployment of batch and streaming jobs is done via the [spark-submit script](https://spark.apache.org/docs/latest/submitting-applications.html). You will need a cluster somewhere, if you want to play locally with a cluster in docker you may be able to tweak [this project](https://github.com/rberenguel/spark_hadoop_kudu) I have in docker-compose, building a stack of HDFS, Spark and Kudu locally.\n",
    "\n",
    "## Testing\n",
    "\n",
    "Always write tests. Use pytest, or your preferred framework, together with a library like [spark-testing-base](https://github.com/holdenk/spark-testing-base/blob/master/python/sparktestingbase/test/helloworld_test.py).\n",
    "\n",
    "spark-testing-base basically handles creating and removing a Spark session when needed, otherwise you can create your own fixtures for pytest (I vastly prefer `pytest` as a testing framework in case it wasn't clear) with a fixture like this one (taken from [here](https://dev.to/diogoaurelio/getting-started-with-spark-part-4----unit-testing-26a5), if you want to see a `unittest` approach, see [here](https://blog.cambridgespark.com/unit-testing-with-pyspark-fb31671b1ad8) and [here](https://medium.com/albert-franzi/how-to-unittest-pyspark-udfs-e34044e6ab68))\n",
    "\n",
    "```python\n",
    "import pytest\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext, SparkConf\n",
    "import logging\n",
    "\n",
    "\n",
    "def quiet_py4j():\n",
    "    \"\"\" turn down spark logging for the test context \"\"\"\n",
    "    logger = logging.getLogger('py4j')\n",
    "    logger.setLevel(logging.WARN)\n",
    "\n",
    "\n",
    "@pytest.fixture(scope=\"session\")\n",
    "def spark_context(request):\n",
    "    \"\"\"\n",
    "        fixture for creating a spark session\n",
    "    Args:\n",
    "        request: pytest.FixtureRequest object\n",
    "    \"\"\"\n",
    "    conf = SparkConf() \\\n",
    "        .setMaster(\"local[2]\") \\\n",
    "        .setAppName(\"pytest-pyspark-local-testing\")\n",
    "    sc = SparkContext(conf=conf)\n",
    "    request.addfinalizer(lambda: sc.stop())\n",
    "    quiet_py4j()\n",
    "    return sc\n",
    "\n",
    "\n",
    "@pytest.fixture(scope=\"session\")\n",
    "def spark_session(request):\n",
    "    \"\"\"\n",
    "        fixture for creating a spark session\n",
    "    Args:\n",
    "        request: pytest.FixtureRequest object\n",
    "    \"\"\"\n",
    "    spark_conf = SparkConf() \\\n",
    "        .setMaster(\"local[2]\") \\\n",
    "        .setAppName(\"pytest-pyspark2.+-local-testing\")\n",
    "    spark = SparkSession.builder.config(conf=spark_conf).getOrCreate()\n",
    "    request.addfinalizer(lambda: spark.stop())\n",
    "    quiet_py4j()\n",
    "    return spark\n",
    "```\n",
    "\n",
    "Use a linting tool as well.\n",
    "\n",
    "## Code structure and naming\n",
    "\n",
    "Spark code is still Python code, mostly, so just use your usual Python common sense. Methods on PySpark sometimes follow Scala naming conventions instead (`camelCase`), if for some reason some method in your code looks better to you in `camelCase`, nobody will scream at you except `pylint`.\n",
    "\n",
    "## Debugging failing stuff\n",
    "\n",
    "- The history server is your friend. There, you can investigate out of memory issues, bad skew of the data and many other things. \n",
    "- Add logging where possible, logging is also your friend.\n",
    "- Most errors in Spark applications are either executor out of memory, driver out of memory or off-heap out of memory. Look up how to increase these parameters and tune your jobs. Once tuned to run, investigate _why_ you need so much.\n",
    "- The JVM stacktraces are not your friend, and you will need to learn to understand them. To train yourself, break spark jobs frequently in testing and locally. \n",
    "\n",
    "### Enjoy!"
   ]
  },
  {
   "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.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}