33.All-Joins-Explained.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Join():\n",
    "                    Syntax: RDD.join(<otherRDD>, numPartitions=None)\n",
    "The join() transformation is an implementation of an inner join, matching two key/value pair\n",
    "RDDs by their key.\n",
    "\n",
    "The optional numPartitions argument determines how many partitions to create in the resultant\n",
    "dataset. If this is not specified, the default value for the spark.default.parallelism\n",
    "configuration parameter is used. The numPartitions argument has the same behavior for other\n",
    "types of join operations in the Spark API as well.\n",
    "\n",
    "The RDD returned is a structure containing the matched key and a value that is a tuple containing\n",
    "all the matched records from both RDDs as a list object. (This is where it may sound a bit foreign to you if you are used to performing INNER JOIN operations in SQL, which returns a flattened list\n",
    "of columns from both entities.)\n",
    "\n",
    "Let's see it through an example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('spark.master', 'local'),\n",
       " ('spark.app.id', 'local-1590555401216'),\n",
       " ('spark.driver.port', '59330'),\n",
       " ('spark.rdd.compress', 'True'),\n",
       " ('spark.app.name', 'All-Joins'),\n",
       " ('spark.serializer.objectStreamReset', '100'),\n",
       " ('spark.executor.id', 'driver'),\n",
       " ('spark.submit.deployMode', 'client'),\n",
       " ('spark.ui.showConsoleProgress', 'true'),\n",
       " ('spark.driver.host', 'DESKTOP-I7971JS')]"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext,SparkConf\n",
    "\n",
    "configure = SparkConf().setAppName(\"All-Joins\").setMaster(\"local\")\n",
    "sc = SparkContext(conf = configure)\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "        .appName(\"All-Joins\") \\\n",
    "        .getOrCreate()\n",
    "    \n",
    "spark.sparkContext.getConf().getAll()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "stores = sc.parallelize(\n",
    "                        [\n",
    "                        (100, 'Boca Raton'),\n",
    "                        (101, 'Columbia'),\n",
    "                        (102, 'Cambridge'),\n",
    "                        (103, 'Naperville')\n",
    "                        ]\n",
    ")\n",
    "# stores schema (store_id, store_location)\n",
    "salespeople = sc.parallelize(\n",
    "                              [\n",
    "                                (1, 'Henry', 100),\n",
    "                                (2, 'Karen', 100),\n",
    "                                (3, 'Paul', 101),\n",
    "                                (4, 'Jimmy', 102),\n",
    "                                (5, 'Janice', None)\n",
    "                              ]\n",
    ")\n",
    "# salespeople schema (salesperson_id, salesperson_name, store_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(100, ((1, 'Henry', 100), 'Boca Raton')),\n",
       " (100, ((2, 'Karen', 100), 'Boca Raton')),\n",
       " (102, ((4, 'Jimmy', 102), 'Cambridge')),\n",
       " (101, ((3, 'Paul', 101), 'Columbia'))]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "joinedDF = salespeople.keyBy(lambda x:x[2]) \\\n",
    "                      .join(stores) \n",
    "\n",
    "joinedDF.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This join() operation returns all salespeople assigned to stores keyed by the store ID (the join\n",
    "key) along with the entire store record and salesperson record. Notice that the resultant RDD\n",
    "contains duplicate data. You could (and should in many cases) follow the join() with a map()\n",
    "transformation to prune fields or project only the fields required for further processing.\n",
    "\n",
    "# Optimizing Joins in Spark:\n",
    "Joins involving RDDs that span more than one partition—and many do—require a shuffle.\n",
    "Spark generally plans and implements this activity to achieve the most optimal performance\n",
    "possible; however, a simple axiom to remember is “join large by small.” This means to reference\n",
    "the large RDD (the one with the most elements, if this is known) first, followed by the\n",
    "smaller of the two RDDs. This will seem strange for users coming from relational database\n",
    "programming backgrounds, but unlike with relational database systems, joins in Spark are relatively\n",
    "inefficient. And unlike with most databases, there are no indexes or statistics to optimize\n",
    "the join, so the optimizations you provide are essential to maximizing performance."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# leftOuterJoin():\n",
    "                Syntax:  RDD.leftOuterJoin(<otherRDD>, numPartitions=None)\n",
    "                \n",
    "The leftOuterJoin() transformation returns all elements or records from the first RDD referenced.\n",
    "If keys from the first (or left) RDD are present in the right RDD, then the right RDD record\n",
    "is returned along with the left RDD record. Otherwise, the right RDD record is None (empty).\n",
    "\n",
    "The example shown below uses the leftOuterJoin() transformation to identify salespeople\n",
    "with no stores."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['salesperson Janice has no store']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "withNoStores = salespeople.keyBy(lambda x: x[2]) \\\n",
    "                          .leftOuterJoin(stores) \\\n",
    "                          .filter(lambda x : x[1][1] is None) \\\n",
    "                          .map(lambda x : \"salesperson \" + x[1][0][1] + \" has no store\")\n",
    "withNoStores.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# rightOuterJoin():\n",
    "                    Syntax:    RDD.rightOuterJoin(<otherRDD>, numPartitions=None)\n",
    "The rightOuterJoin() transformation returns all elements or records from the second RDD\n",
    "referenced. If keys from the second (or right) RDD are present in the left RDD, then the left\n",
    "RDD record is returned along with the right RDD record. Otherwise, the left RDD record is None\n",
    "(empty).\n",
    "\n",
    "The section below shows an example of how the rightOuterJoin() transformation can be used to\n",
    "identify stores with no salespeople."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Naperville Store has no saleperson']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "storeWithoutSalesPeople = salespeople.keyBy(lambda x : x[2]) \\\n",
    "                                     .rightOuterJoin(stores) \\\n",
    "                                     .filter(lambda x: x[1][0] is None) \\\n",
    "                                     .map(lambda x: x[1][1] + \" Store has no saleperson\")\n",
    "storeWithoutSalesPeople.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# fullOuterJoin():\n",
    "                            Syntax:  RDD.fullOuterJoin(<otherRDD>, numPartitions=None)\n",
    "The fullOuterJoin() transforms all elements from both RDDs whether there is a key matched\n",
    "or not. Keys not matched from either the left or right dataset are represented as None (empty).\n",
    "\n",
    "The section below shows an example of how the fullOuterJoin() transformation can be used to identify\n",
    "stores with no salespeople as well as salespeople with no stores."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(None, ((5, 'Janice', None), None)), (103, (None, 'Naperville'))]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "outerJoin = salespeople.keyBy(lambda x:x[2]) \\\n",
    "                        .fullOuterJoin(stores) \\\n",
    "                        .filter(lambda x: x[1][0] is None or x[1][1] is None) \n",
    "                        \n",
    "outerJoin.collect()"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}