PySpark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark SQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The pyspark.sql module"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Important classes of Spark SQL and DataFrames:\n",
    "\n",
    "* `pyspark.sql.SparkSession` Main entry point for DataFrame and SQL functionality.\n",
    "\n",
    "* `pyspark.sql.DataFrame` A distributed collection of data grouped into named columns.\n",
    "\n",
    "* `pyspark.sql.Column` A column expression in a DataFrame.\n",
    "\n",
    "* `pyspark.sql.Row` A row of data in a DataFrame.\n",
    "\n",
    "* `pyspark.sql.GroupedData` Aggregation methods, returned by DataFrame.groupBy().\n",
    "\n",
    "* `pyspark.sql.DataFrameNaFunctions` Methods for handling missing data (null values).\n",
    "\n",
    "* `pyspark.sql.DataFrameStatFunctions` Methods for statistics functionality.\n",
    "\n",
    "* `pyspark.sql.functions` List of built-in functions available for DataFrame.\n",
    "\n",
    "* `pyspark.sql.types` List of data types available.\n",
    "\n",
    "* `pyspark.sql.Window` For working with window functions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "http://spark.apache.org/docs/2.2.0/api/python/pyspark.sql.html\n",
    "\n",
    "https://spark.apache.org/docs/2.2.0/sql-programming-guide.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. SparkSession\n",
    "\n",
    "The traditional way to interact with Spark is the SparkContext. In the notebooks we get that from the pyspark driver.\n",
    "\n",
    "From 2.0 we can use SparkSession to replace SparkConf, SparkContext and SQLContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<pyspark.sql.session.SparkSession object at 0x7f1f730a0e48>\n"
     ]
    }
   ],
   "source": [
    "# Import SparkSession from pyspark.sql\n",
    "from pyspark.sql import SparkSession\n",
    "# Create a Session\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "# How is my session?\n",
    "print(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://10.0.2.15:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>pyspark-shell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f1f730a0e48>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Passing other options to spark session:\n",
    "    \n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.config('someoption.key','somevalue').getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can check option values in the resulting session like this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('someoption.key', 'somevalue'),\n",
       " ('spark.rdd.compress', 'True'),\n",
       " ('spark.driver.port', '38059'),\n",
       " ('spark.serializer.objectStreamReset', '100'),\n",
       " ('spark.master', 'local[*]'),\n",
       " ('spark.executor.id', 'driver'),\n",
       " ('spark.submit.deployMode', 'client'),\n",
       " ('spark.driver.host', '10.0.2.15'),\n",
       " ('spark.app.id', 'local-1558192457954'),\n",
       " ('spark.app.name', 'pyspark-shell'),\n",
       " ('spark.ui.showConsoleProgress', 'true')]"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sparkContext.getConf().getAll()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Creating DataFrames\n",
    "\n",
    "SparkSession.createDataFrame: from a RDD, a list or a pandas.DataFrame."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1 Zip "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['manager', 'mechanic', 'mechanic', 'manager', 'sales']\n",
      "<zip object at 0x7f1f67052548>\n"
     ]
    }
   ],
   "source": [
    "# Create the rows for the Dataframe\n",
    "import random\n",
    "random.seed(42)\n",
    "ids= range(5)\n",
    "positions=[random.choice(['mechanic','sales','manager']) for id_ in ids]\n",
    "print(positions)\n",
    "# Join ids with positions through zip command\n",
    "rows = zip(ids,positions)\n",
    "print(rows)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the Dataframe\n",
    "df = spark.createDataFrame(rows)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "| _1|      _2|\n",
      "+---+--------+\n",
      "|  0| manager|\n",
      "|  1|mechanic|\n",
      "|  2|mechanic|\n",
      "|  3| manager|\n",
      "|  4|   sales|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Show the content. It's similar to .head() in Pandas \n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(_1=0, _2='manager'),\n",
       " Row(_1=1, _2='mechanic'),\n",
       " Row(_1=2, _2='mechanic'),\n",
       " Row(_1=3, _2='manager'),\n",
       " Row(_1=4, _2='sales')]"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Collect is another way to see the Dataframe\n",
    "df.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.2 Row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "list"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import Row\n",
    "\n",
    "rows = [Row(id=id_, position=position_) for id_, position_ in zip(ids,positions)]\n",
    "type(rows)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "| id|position|\n",
      "+---+--------+\n",
      "|  0| manager|\n",
      "|  1|mechanic|\n",
      "|  2|mechanic|\n",
      "|  3| manager|\n",
      "|  4|   sales|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.createDataFrame(rows)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.3 Read"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```python\n",
    "# Read a file from a CSV\n",
    "df = spark.read.csv(file_path,header=True, inferSchema=True)\n",
    "# Read a file from a CSV\n",
    "df = spark.read.format('csv').options(header=True, inferSchema=True).load(file_path)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: long (nullable = true)\n",
      " |-- _c1: integer (nullable = true)\n",
      " |-- _c2: string (nullable = true)\n",
      " |-- _c3: string (nullable = true)\n",
      " |-- _c4: string (nullable = true)\n",
      " |-- _c5: string (nullable = true)\n",
      " |-- _c6: double (nullable = true)\n",
      " |-- _c7: string (nullable = true)\n",
      " |-- _c8: integer (nullable = true)\n",
      " |-- _c9: string (nullable = true)\n",
      " |-- _c10: string (nullable = true)\n",
      " |-- _c11: string (nullable = true)\n",
      " |-- _c12: integer (nullable = true)\n",
      " |-- _c13: string (nullable = true)\n",
      " |-- _c14: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Infer Schema\n",
    "df = spark.read.csv(path,header=False,inferSchema=True)\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: string (nullable = true)\n",
      " |-- _c1: string (nullable = true)\n",
      " |-- _c2: string (nullable = true)\n",
      " |-- _c3: string (nullable = true)\n",
      " |-- _c4: string (nullable = true)\n",
      " |-- _c5: string (nullable = true)\n",
      " |-- _c6: string (nullable = true)\n",
      " |-- _c7: string (nullable = true)\n",
      " |-- _c8: string (nullable = true)\n",
      " |-- _c9: string (nullable = true)\n",
      " |-- _c10: string (nullable = true)\n",
      " |-- _c11: string (nullable = true)\n",
      " |-- _c12: string (nullable = true)\n",
      " |-- _c13: string (nullable = true)\n",
      " |-- _c14: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# All our columns are strings by default\n",
    "path = '/home/eduardo/Repos/Master_DataScience/19_Spark/coupon150720.csv'\n",
    "df = spark.read.csv(path,header=False).limit(10)\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Inferring and Specifying Schemas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+------+\n",
      "|           _c0|_c1|_c2|   _c6|\n",
      "+--------------+---+---+------+\n",
      "|79062005698500|  1|MAA| 56.79|\n",
      "|79062005698500|  2|AUH| 84.34|\n",
      "|79062005924069|  1|CJB|  60.0|\n",
      "|79065668570385|  1|DEL|160.63|\n",
      "|79065668737021|  1|AUH|152.46|\n",
      "+--------------+---+---+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = df.select('_c0','_c1','_c2','_c6')\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: string (nullable = true)\n",
      " |-- _c1: string (nullable = true)\n",
      " |-- _c2: string (nullable = true)\n",
      " |-- _c6: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: string (nullable = true)\n",
      " |-- _c1: integer (nullable = true)\n",
      " |-- _c2: string (nullable = true)\n",
      " |-- _c6: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import types\n",
    "\n",
    "# Column Types\n",
    "fields = [types.StructField('_c0',types.StringType()),\n",
    "          types.StructField('_c1',types.IntegerType()),\n",
    "          types.StructField('_c2',types.StringType()),\n",
    "          types.StructField('_c6',types.FloatType())]\n",
    "\n",
    "# Create my Schema\n",
    "my_schema = types.StructType(fields)\n",
    "\n",
    "# Read Dataframe\n",
    "df = spark.read.csv(path, schema=my_schema).select('_c0','_c1','_c2','_c6')\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Filtering and Selecting"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.1 Selecting"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|           _c0|\n",
      "+--------------+\n",
      "|79062005698500|\n",
      "|79062005698500|\n",
      "|79062005924069|\n",
      "|79065668570385|\n",
      "|79065668737021|\n",
      "+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('_c0').show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|           _c0|\n",
      "+--------------+\n",
      "|79062005698500|\n",
      "|79062005698500|\n",
      "|79062005924069|\n",
      "|79065668570385|\n",
      "|79065668737021|\n",
      "+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df['_c0']).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|           _c0|\n",
      "+--------------+\n",
      "|79062005698500|\n",
      "|79062005698500|\n",
      "|79062005924069|\n",
      "|79065668570385|\n",
      "|79065668737021|\n",
      "+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import functions as f\n",
    "\n",
    "df.select(f.col('_c0')).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|           _c0|\n",
      "+--------------+\n",
      "|79062005698500|\n",
      "|79062005698500|\n",
      "|79062005924069|\n",
      "|79065668570385|\n",
      "|79065668737021|\n",
      "+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df._c0).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.2 Filtering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "|           _c0|_c1|_c2|_c3|_c4|_c5|   _c6|_c7|_c8|_c9|_c10|_c11|  _c12|_c13|_c14|\n",
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "|79062005698500|  2|AUH|CDG| 9W| 9W| 84.34|USD|  1|  H|   H|6120|150905|  OK|IAF0|\n",
      "|79065668737021|  1|AUH|IXE| 9W| 9W|152.46|USD|  1|  V|   V|0501|150803|  OK|INA0|\n",
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Read CSV\n",
    "df = spark.read.csv(path,header=False,inferSchema=True).limit(10)\n",
    "# Filter\n",
    "filtered = df.filter(df['_c2'] == 'AUH')\n",
    "filtered.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "|           _c0|_c1|_c2|_c3|_c4|_c5|   _c6|_c7|_c8|_c9|_c10|_c11|  _c12|_c13|_c14|\n",
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "|79062005698500|  1|MAA|AUH| 9W| 9W| 56.79|USD|  1|  H|   H|0526|150904|  OK|IAF0|\n",
      "|79062005698500|  2|AUH|CDG| 9W| 9W| 84.34|USD|  1|  H|   H|6120|150905|  OK|IAF0|\n",
      "|79062005924069|  1|CJB|MAA| 9W| 9W|  60.0|USD|  1|  H|   H|2768|150721|  OK|IAA0|\n",
      "|79065668570385|  1|DEL|DXB| 9W| 9W|160.63|USD|  2|  S|   S|0546|150804|  OK|INA0|\n",
      "|79065668737021|  1|AUH|IXE| 9W| 9W|152.46|USD|  1|  V|   V|0501|150803|  OK|INA0|\n",
      "|79062006192650|  1|RPR|BOM| 9W| 9W|  68.5|USD|  1|  K|   K|2202|150720|  OK|IAE0|\n",
      "|79062006192650|  2|BOM|RPR| 9W| 9W|  68.5|USD|  1|  H|   H|0377|150721|  OK|IAE0|\n",
      "|79062005733853|  1|DEL|DED| 9W| 9W| 56.16|USD|  1|  V|   V|2839|150801|  OK|INA0|\n",
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filter\n",
    "filtered2 = df.filter(df['_c6'] > 50)\n",
    "filtered2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "|           _c0|_c1|_c2|_c3|_c4|_c5|   _c6|_c7|_c8|_c9|_c10|_c11|  _c12|_c13|_c14|\n",
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "|79065668570385|  1|DEL|DXB| 9W| 9W|160.63|USD|  2|  S|   S|0546|150804|  OK|INA0|\n",
      "+--------------+---+---+---+---+---+------+---+---+---+----+----+------+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Where is similar to Filter\n",
    "whered = df.where(f.col('_c8') >= 2)\n",
    "whered.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Adding Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+-------+\n",
      "|           _c0|_c1|_c2|_c3| _c4| _c5|   _c6|_c7|_c8| _c9|_c10|_c11|  _c12|_c13|_c14|  price|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+-------+\n",
      "|79062005698500|  1|MAA|AUH|  9W|  9W| 56.79|USD|  1|   H|   H|0526|150904|  OK|IAF0| 5679.0|\n",
      "|79062005698500|  2|AUH|CDG|  9W|  9W| 84.34|USD|  1|   H|   H|6120|150905|  OK|IAF0| 8434.0|\n",
      "|79062005924069|  1|CJB|MAA|  9W|  9W|  60.0|USD|  1|   H|   H|2768|150721|  OK|IAA0| 6000.0|\n",
      "|79065668570385|  1|DEL|DXB|  9W|  9W|160.63|USD|  2|   S|   S|0546|150804|  OK|INA0|16063.0|\n",
      "|79065668737021|  1|AUH|IXE|  9W|  9W|152.46|USD|  1|   V|   V|0501|150803|  OK|INA0|15246.0|\n",
      "|79062006192650|  1|RPR|BOM|  9W|  9W|  68.5|USD|  1|   K|   K|2202|150720|  OK|IAE0| 6850.0|\n",
      "|79062006192650|  2|BOM|RPR|  9W|  9W|  68.5|USD|  1|   H|   H|0377|150721|  OK|IAE0| 6850.0|\n",
      "|79062005733853|  1|DEL|DED|  9W|  9W| 56.16|USD|  1|   V|   V|2839|150801|  OK|INA0| 5616.0|\n",
      "|79062005836987|  1|ATL|LGA|  AA|  AA|  28.3|USD|  1|   V|   V|3237|150903|  OK|INB0| 2830.0|\n",
      "|79062005836987|  2|LGA|EWR|null|null|   0.0|USD|  1|null|null|VOID|  null|null|INA0|    0.0|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn('price',f.col('_c6')*100).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+\n",
      "|           _c0|_c1|_c2|_c3| _c4| _c5|   _c6|_c7|_c8| _c9|_c10|_c11|  _c12|_c13|_c14|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+\n",
      "|79062005698500|  1|MAA|AUH|  9W|  9W| 56.79|USD|  1|   H|   H|0526|150904|  OK|IAF0|\n",
      "|79062005698500|  2|AUH|CDG|  9W|  9W| 84.34|USD|  1|   H|   H|6120|150905|  OK|IAF0|\n",
      "|79062005924069|  1|CJB|MAA|  9W|  9W|  60.0|USD|  1|   H|   H|2768|150721|  OK|IAA0|\n",
      "|79065668570385|  1|DEL|DXB|  9W|  9W|160.63|USD|  2|   S|   S|0546|150804|  OK|INA0|\n",
      "|79065668737021|  1|AUH|IXE|  9W|  9W|152.46|USD|  1|   V|   V|0501|150803|  OK|INA0|\n",
      "|79062006192650|  1|RPR|BOM|  9W|  9W|  68.5|USD|  1|   K|   K|2202|150720|  OK|IAE0|\n",
      "|79062006192650|  2|BOM|RPR|  9W|  9W|  68.5|USD|  1|   H|   H|0377|150721|  OK|IAE0|\n",
      "|79062005733853|  1|DEL|DED|  9W|  9W| 56.16|USD|  1|   V|   V|2839|150801|  OK|INA0|\n",
      "|79062005836987|  1|ATL|LGA|  AA|  AA|  28.3|USD|  1|   V|   V|3237|150903|  OK|INB0|\n",
      "|79062005836987|  2|LGA|EWR|null|null|   0.0|USD|  1|null|null|VOID|  null|null|INA0|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = df.withColumn('_c1', f.col('_c1').cast(types.IntegerType()))\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. User Defined Functions UDF\n",
    "\n",
    "There are many useful functions in pyspark.sql.functions. These work on columns, that is, they are vectorial.\n",
    "\n",
    "We can write User Defined Functions (udfs), which allow us to \"vectorize\" operations: write a standard function to process single elements, then build a udf with that that works on columns in a DataFrame, like a SQL function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.0"
      ]
     },
     "execution_count": 95,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import math\n",
    "math.log1p(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [],
   "source": [
    "# df.select(math.log1p('_c1')) this is going to return an error\n",
    "# we need to define an UDF to do this"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create Function and Define UDF\n",
    "def log1p_fun(num):\n",
    "    return math.log1p(num)\n",
    "\n",
    "udf_log1p = f.udf(log1p_fun)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+------+------------------+\n",
      "|           _c0|_c2|_c3|   _c6|      newColumnUDF|\n",
      "+--------------+---+---+------+------------------+\n",
      "|79062005698500|MAA|AUH| 56.79|0.6931471805599453|\n",
      "|79062005698500|AUH|CDG| 84.34|1.0986122886681096|\n",
      "|79062005924069|CJB|MAA|  60.0|0.6931471805599453|\n",
      "|79065668570385|DEL|DXB|160.63|0.6931471805599453|\n",
      "|79065668737021|AUH|IXE|152.46|0.6931471805599453|\n",
      "|79062006192650|RPR|BOM|  68.5|0.6931471805599453|\n",
      "|79062006192650|BOM|RPR|  68.5|1.0986122886681096|\n",
      "|79062005733853|DEL|DED| 56.16|0.6931471805599453|\n",
      "|79062005836987|ATL|LGA|  28.3|0.6931471805599453|\n",
      "|79062005836987|LGA|EWR|   0.0|1.0986122886681096|\n",
      "+--------------+---+---+------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn('newColumnUDF',udf_log1p(f.col('_c1'))).select('_c0','_c2','_c3','_c6','newColumnUDF').show()    # f.col('_c1') == df['_c1']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+------+----------------+\n",
      "|           _c0|_c2|_c3|   _c6|newColOddLetters|\n",
      "+--------------+---+---+------+----------------+\n",
      "|79062005698500|MAA|AUH| 56.79|              MA|\n",
      "|79062005698500|AUH|CDG| 84.34|              AH|\n",
      "|79062005924069|CJB|MAA|  60.0|              CB|\n",
      "|79065668570385|DEL|DXB|160.63|              DL|\n",
      "|79065668737021|AUH|IXE|152.46|              AH|\n",
      "|79062006192650|RPR|BOM|  68.5|              RR|\n",
      "|79062006192650|BOM|RPR|  68.5|              BM|\n",
      "|79062005733853|DEL|DED| 56.16|              DL|\n",
      "|79062005836987|ATL|LGA|  28.3|              AL|\n",
      "|79062005836987|LGA|EWR|   0.0|              LA|\n",
      "+--------------+---+---+------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# UDF to get odd letters\n",
    "def odd_letters(word):\n",
    "    return word[::2]\n",
    "udf_odd = f.udf(odd_letters, returnType=types.StringType()) # It will be StringType by default\n",
    "\n",
    "df.withColumn('newColOddLetters',udf_odd(df['_c2'])).select('_c0','_c2','_c3','_c6','newColOddLetters').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we want the resulting columns to be of a particular type, we need to specify the return type. This is because in Python return types can not be inferred."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+------+---------+\n",
      "|           _c0|_c2|_c3|   _c6|newColUdf|\n",
      "+--------------+---+---+------+---------+\n",
      "|79062005698500|MAA|AUH| 56.79|0.6931472|\n",
      "|79062005698500|AUH|CDG| 84.34|1.0986123|\n",
      "|79062005924069|CJB|MAA|  60.0|0.6931472|\n",
      "|79065668570385|DEL|DXB|160.63|0.6931472|\n",
      "|79065668737021|AUH|IXE|152.46|0.6931472|\n",
      "|79062006192650|RPR|BOM|  68.5|0.6931472|\n",
      "|79062006192650|BOM|RPR|  68.5|1.0986123|\n",
      "|79062005733853|DEL|DED| 56.16|0.6931472|\n",
      "|79062005836987|ATL|LGA|  28.3|0.6931472|\n",
      "|79062005836987|LGA|EWR|   0.0|1.0986123|\n",
      "+--------------+---+---+------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "udf_log1p_typed = f.udf(math.log1p, types.FloatType())\n",
    "df.withColumn('newColUdf', udf_log1p_typed(df['_c1'])).select('_c0','_c2','_c3','_c6','newColUdf').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<function __main__.<lambda>(x)>"
      ]
     },
     "execution_count": 107,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f.udf(lambda x: -x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+------+---------+\n",
      "|           _c0|_c2|_c3|   _c6|   prices|\n",
      "+--------------+---+---+------+---------+\n",
      "|79062005698500|MAA|AUH| 56.79|    Cheap|\n",
      "|79062005698500|AUH|CDG| 84.34|    Cheap|\n",
      "|79062005924069|CJB|MAA|  60.0|    Cheap|\n",
      "|79065668570385|DEL|DXB|160.63|Expensive|\n",
      "|79065668737021|AUH|IXE|152.46|Expensive|\n",
      "|79062006192650|RPR|BOM|  68.5|    Cheap|\n",
      "|79062006192650|BOM|RPR|  68.5|    Cheap|\n",
      "|79062005733853|DEL|DED| 56.16|    Cheap|\n",
      "|79062005836987|ATL|LGA|  28.3|    Cheap|\n",
      "|79062005836987|LGA|EWR|   0.0|    Cheap|\n",
      "+--------------+---+---+------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def prices(num):\n",
    "    if num > 100:\n",
    "        return 'Expensive'\n",
    "    else:\n",
    "        return 'Cheap'\n",
    "    \n",
    "udf_prices = f.udf(prices, types.StringType())\n",
    "\n",
    "df.withColumn('prices',udf_prices(f.col('_c6'))).select('_c0','_c2','_c3','_c6','prices').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. Grouping"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "|_c2|avg(_c6)|\n",
      "+---+--------+\n",
      "|MAA|   56.79|\n",
      "|AUH|   118.4|\n",
      "|CJB|    60.0|\n",
      "|DEL| 108.395|\n",
      "|RPR|    68.5|\n",
      "|BOM|    68.5|\n",
      "|ATL|    28.3|\n",
      "|LGA|     0.0|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('_c2').mean('_c6').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+------+-----+\n",
      "|_c2|average|   std|count|\n",
      "+---+-------+------+-----+\n",
      "|MAA|  56.79|   0.0|    1|\n",
      "|AUH|  118.4| 34.06|    2|\n",
      "|CJB|   60.0|   0.0|    1|\n",
      "|DEL|108.395|52.235|    2|\n",
      "|RPR|   68.5|   0.0|    1|\n",
      "|BOM|   68.5|   0.0|    1|\n",
      "|ATL|   28.3|   0.0|    1|\n",
      "|LGA|    0.0|   0.0|    1|\n",
      "+---+-------+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "stats = df.groupBy('_c2').agg(f.mean('_c6').alias('average'),\n",
    "                              f.stddev_pop('_c6').alias('std'),\n",
    "                              f.count('_c6').alias('count')\n",
    "                              ).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+-------+\n",
      "|_c2|_c8|average|\n",
      "+---+---+-------+\n",
      "|MAA|  1|  56.79|\n",
      "|AUH|  1|  118.4|\n",
      "|CJB|  1|   60.0|\n",
      "|DEL|  2| 160.63|\n",
      "|RPR|  1|   68.5|\n",
      "|BOM|  1|   68.5|\n",
      "|DEL|  1|  56.16|\n",
      "|ATL|  1|   28.3|\n",
      "|LGA|  1|    0.0|\n",
      "+---+---+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(['_c2','_c8']).agg(f.mean('_c6').alias('average')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 8. Joins"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------+\n",
      "|code|      name|\n",
      "+----+----------+\n",
      "| MAA|   Chennai|\n",
      "| AUH|  Abu Dabi|\n",
      "| CJB|Coimbatore|\n",
      "| DEL|     Delhi|\n",
      "| RPR|    Raipur|\n",
      "| BOM|    Bombai|\n",
      "| ATL|   Atlanta|\n",
      "| LGA|La Guardia|\n",
      "+----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valuesA = [('MAA','Chennai'),('AUH','Abu Dabi'),('CJB','Coimbatore'),('DEL','Delhi'),('RPR','Raipur'),('BOM','Bombai'),('ATL','Atlanta'),('LGA','La Guardia')]\n",
    "TableA = spark.createDataFrame(valuesA,['code','name'])\n",
    "TableA.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+----+----------+\n",
      "|           _c0|_c1|_c2|_c3| _c4| _c5|   _c6|_c7|_c8| _c9|_c10|_c11|  _c12|_c13|_c14|code|      name|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+----+----------+\n",
      "|79062005698500|  1|MAA|AUH|  9W|  9W| 56.79|USD|  1|   H|   H|0526|150904|  OK|IAF0| MAA|   Chennai|\n",
      "|79065668737021|  1|AUH|IXE|  9W|  9W|152.46|USD|  1|   V|   V|0501|150803|  OK|INA0| AUH|  Abu Dabi|\n",
      "|79062005698500|  2|AUH|CDG|  9W|  9W| 84.34|USD|  1|   H|   H|6120|150905|  OK|IAF0| AUH|  Abu Dabi|\n",
      "|79062005924069|  1|CJB|MAA|  9W|  9W|  60.0|USD|  1|   H|   H|2768|150721|  OK|IAA0| CJB|Coimbatore|\n",
      "|79062005733853|  1|DEL|DED|  9W|  9W| 56.16|USD|  1|   V|   V|2839|150801|  OK|INA0| DEL|     Delhi|\n",
      "|79065668570385|  1|DEL|DXB|  9W|  9W|160.63|USD|  2|   S|   S|0546|150804|  OK|INA0| DEL|     Delhi|\n",
      "|79062006192650|  1|RPR|BOM|  9W|  9W|  68.5|USD|  1|   K|   K|2202|150720|  OK|IAE0| RPR|    Raipur|\n",
      "|79062006192650|  2|BOM|RPR|  9W|  9W|  68.5|USD|  1|   H|   H|0377|150721|  OK|IAE0| BOM|    Bombai|\n",
      "|79062005836987|  1|ATL|LGA|  AA|  AA|  28.3|USD|  1|   V|   V|3237|150903|  OK|INB0| ATL|   Atlanta|\n",
      "|79062005836987|  2|LGA|EWR|null|null|   0.0|USD|  1|null|null|VOID|  null|null|INA0| LGA|La Guardia|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.join(TableA, on=df['_c2']==TableA['code']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+---+---+-----+---+---+---+----+----+------+----+----+----+----------+\n",
      "|           _c0|_c1|_c2|_c3|_c4|_c5|  _c6|_c7|_c8|_c9|_c10|_c11|  _c12|_c13|_c14|code|      name|\n",
      "+--------------+---+---+---+---+---+-----+---+---+---+----+----+------+----+----+----+----------+\n",
      "|79062005924069|  1|CJB|MAA| 9W| 9W| 60.0|USD|  1|  H|   H|2768|150721|  OK|IAA0| MAA|   Chennai|\n",
      "|79062005698500|  1|MAA|AUH| 9W| 9W|56.79|USD|  1|  H|   H|0526|150904|  OK|IAF0| AUH|  Abu Dabi|\n",
      "|79062006192650|  2|BOM|RPR| 9W| 9W| 68.5|USD|  1|  H|   H|0377|150721|  OK|IAE0| RPR|    Raipur|\n",
      "|79062006192650|  1|RPR|BOM| 9W| 9W| 68.5|USD|  1|  K|   K|2202|150720|  OK|IAE0| BOM|    Bombai|\n",
      "|79062005836987|  1|ATL|LGA| AA| AA| 28.3|USD|  1|  V|   V|3237|150903|  OK|INB0| LGA|La Guardia|\n",
      "+--------------+---+---+---+---+---+-----+---+---+---+----+----+------+----+----+----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.join(TableA, on=df['_c3']==TableA['code']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+----+----------+\n",
      "|           _c0|_c1|_c2|_c3| _c4| _c5|   _c6|_c7|_c8| _c9|_c10|_c11|  _c12|_c13|_c14|code|      name|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+----+----------+\n",
      "|79062005698500|  1|MAA|AUH|  9W|  9W| 56.79|USD|  1|   H|   H|0526|150904|  OK|IAF0| AUH|  Abu Dabi|\n",
      "|79062005698500|  2|AUH|CDG|  9W|  9W| 84.34|USD|  1|   H|   H|6120|150905|  OK|IAF0|null|      null|\n",
      "|79062005836987|  1|ATL|LGA|  AA|  AA|  28.3|USD|  1|   V|   V|3237|150903|  OK|INB0| LGA|La Guardia|\n",
      "|79062006192650|  1|RPR|BOM|  9W|  9W|  68.5|USD|  1|   K|   K|2202|150720|  OK|IAE0| BOM|    Bombai|\n",
      "|79065668570385|  1|DEL|DXB|  9W|  9W|160.63|USD|  2|   S|   S|0546|150804|  OK|INA0|null|      null|\n",
      "|79062005836987|  2|LGA|EWR|null|null|   0.0|USD|  1|null|null|VOID|  null|null|INA0|null|      null|\n",
      "|79062005733853|  1|DEL|DED|  9W|  9W| 56.16|USD|  1|   V|   V|2839|150801|  OK|INA0|null|      null|\n",
      "|79062006192650|  2|BOM|RPR|  9W|  9W|  68.5|USD|  1|   H|   H|0377|150721|  OK|IAE0| RPR|    Raipur|\n",
      "|79062005924069|  1|CJB|MAA|  9W|  9W|  60.0|USD|  1|   H|   H|2768|150721|  OK|IAA0| MAA|   Chennai|\n",
      "|79065668737021|  1|AUH|IXE|  9W|  9W|152.46|USD|  1|   V|   V|0501|150803|  OK|INA0|null|      null|\n",
      "+--------------+---+---+---+----+----+------+---+---+----+----+----+------+----+----+----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.join(TableA, on=df['_c3']==TableA['code'],how='left').show()"
   ]
  }
 ],
 "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}