Spark Processing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('dataframeTesT').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "columns = [\"language\",\"users_count\"]\n",
    "data = [(\"Java\", 20000), (\"Python\", 100000), (\"Scala\", 3000)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd = spark.sparkContext.parallelize(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfFromRDD1 = rdd.toDF()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "columns = [\"language\",\"users_count\"]\n",
    "dfFromRDD1 = rdd.toDF(columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- language: string (nullable = true)\n",
      " |-- users_count: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfFromRDD1.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***Convert PySpark Dataframe to Pandas*** <br>\n",
    "PySpark DataFrame provides a method toPandas() to convert it to Python Pandas DataFrame.<br><br>\n",
    "**toPandas()** results in the collection of all records in the PySpark DataFrame to the driver program and should be done only on a small subset of the data. running on larger dataset’s results in memory error and crashes the application. To deal with a larger dataset, you can also try increasing memory on the driver."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>language</th>\n",
       "      <th>users_count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Java</td>\n",
       "      <td>20000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Python</td>\n",
       "      <td>100000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Scala</td>\n",
       "      <td>3000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  language  users_count\n",
       "0     Java        20000\n",
       "1   Python       100000\n",
       "2    Scala         3000"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dfFromRDD1.toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***Access a dataframe column***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|users_count|\n",
      "+-----------+\n",
      "|      20000|\n",
      "|     100000|\n",
      "|       3000|\n",
      "+-----------+\n",
      "\n",
      "+-----------+\n",
      "|users_count|\n",
      "+-----------+\n",
      "|      20000|\n",
      "|     100000|\n",
      "|       3000|\n",
      "+-----------+\n",
      "\n",
      "+-----------+\n",
      "|users_count|\n",
      "+-----------+\n",
      "|      20000|\n",
      "|     100000|\n",
      "|       3000|\n",
      "+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using DataFrame object (df)\n",
    "dfFromRDD1.select(dfFromRDD1.users_count).show()\n",
    "dfFromRDD1.select(dfFromRDD1[\"users_count\"]).show()\n",
    "\n",
    "#Using SQL col() function\n",
    "from pyspark.sql.functions import col\n",
    "dfFromRDD1.select(col(\"users_count\")).show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "data=[(\"James\",\"Bond\",\"100\",None),\n",
    "      (\"Ann\",\"Varsa\",\"200\",'F'),\n",
    "      (\"Tom Cruise\",\"XXX\",\"400\",''),\n",
    "      (\"Tom Brand\",None,\"400\",'M')] \n",
    "columns=[\"fname\",\"lname\",\"id\",\"gender\"]\n",
    "df=spark.createDataFrame(data,columns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***alias() – Set’s name to Column***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+\n",
      "|first_name|last_name|\n",
      "+----------+---------+\n",
      "|     James|     Bond|\n",
      "|       Ann|    Varsa|\n",
      "|Tom Cruise|      XXX|\n",
      "| Tom Brand|     null|\n",
      "+----------+---------+\n",
      "\n",
      "+--------------+\n",
      "|      fullName|\n",
      "+--------------+\n",
      "|    James,Bond|\n",
      "|     Ann,Varsa|\n",
      "|Tom Cruise,XXX|\n",
      "|          null|\n",
      "+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import expr,lit\n",
    "df.select(df.fname.alias(\"first_name\"), \\\n",
    "          df.lname.alias(\"last_name\")\n",
    "   ).show()\n",
    "\n",
    "#Another example\n",
    "df.select(expr(\" fname ||','|| lname\").alias(\"fullName\") \\\n",
    "   ).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***asc() & desc() – Sort the DataFrame columns by Ascending or Descending order***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+---+------+\n",
      "|     fname|lname| id|gender|\n",
      "+----------+-----+---+------+\n",
      "|       Ann|Varsa|200|     F|\n",
      "|     James| Bond|100|  null|\n",
      "| Tom Brand| null|400|     M|\n",
      "|Tom Cruise|  XXX|400|      |\n",
      "+----------+-----+---+------+\n",
      "\n",
      "+----------+-----+---+------+\n",
      "|     fname|lname| id|gender|\n",
      "+----------+-----+---+------+\n",
      "|Tom Cruise|  XXX|400|      |\n",
      "| Tom Brand| null|400|     M|\n",
      "|     James| Bond|100|  null|\n",
      "|       Ann|Varsa|200|     F|\n",
      "+----------+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.sort(df.fname.asc()).show()\n",
    "df.sort(df.fname.desc()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***cast() & astype() – Used to convert the data Type.***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- fname: string (nullable = true)\n",
      " |-- id: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.fname,df.id.cast(\"int\")).printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***between() – Returns a Boolean expression when a column values in between lower and upper bound.***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+---+------+\n",
      "|fname|lname| id|gender|\n",
      "+-----+-----+---+------+\n",
      "|James| Bond|100|  null|\n",
      "|  Ann|Varsa|200|     F|\n",
      "+-----+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df.id.between(100,300)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***contains() – Checks if a DataFrame column value contains a a value specified in this function.***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+---+------+\n",
      "|     fname|lname| id|gender|\n",
      "+----------+-----+---+------+\n",
      "|Tom Cruise|  XXX|400|      |\n",
      "+----------+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df.fname.contains(\"Cruise\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***startswith() & endswith() – Checks if the value of the DataFrame Column starts and ends with a String respectively.***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+---+------+\n",
      "|     fname|lname| id|gender|\n",
      "+----------+-----+---+------+\n",
      "|Tom Cruise|  XXX|400|      |\n",
      "| Tom Brand| null|400|     M|\n",
      "+----------+-----+---+------+\n",
      "\n",
      "+----------+-----+---+------+\n",
      "|     fname|lname| id|gender|\n",
      "+----------+-----+---+------+\n",
      "|Tom Cruise|  XXX|400|      |\n",
      "+----------+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df.fname.startswith(\"T\")).show()\n",
    "df.filter(df.fname.endswith(\"Cruise\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***isNull & isNotNull() – Checks if the DataFrame column has NULL or non NULL values.***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----+---+------+\n",
      "|    fname|lname| id|gender|\n",
      "+---------+-----+---+------+\n",
      "|Tom Brand| null|400|     M|\n",
      "+---------+-----+---+------+\n",
      "\n",
      "+----------+-----+---+------+\n",
      "|     fname|lname| id|gender|\n",
      "+----------+-----+---+------+\n",
      "|     James| Bond|100|  null|\n",
      "|       Ann|Varsa|200|     F|\n",
      "|Tom Cruise|  XXX|400|      |\n",
      "+----------+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df.lname.isNull()).show()\n",
    "df.filter(df.lname.isNotNull()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***like() & rlike() – Similar to SQL LIKE expression***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+---+\n",
      "|     fname|lname| id|\n",
      "+----------+-----+---+\n",
      "|Tom Cruise|  XXX|400|\n",
      "| Tom Brand| null|400|\n",
      "+----------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.fname,df.lname,df.id) \\\n",
    "  .filter(df.fname.like(\"%om%\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***substr() – Returns a Column after getting sub string from the Column***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|substr|\n",
      "+------+\n",
      "|   Jam|\n",
      "|   Ann|\n",
      "|   Tom|\n",
      "|   Tom|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.fname.substr(1,3).alias(\"substr\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***when() & otherwise() – It is similar to SQL Case When, executes sequence of expressions until it matches the condition and returns a value when match.***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+----------+\n",
      "|     fname|lname|new_gender|\n",
      "+----------+-----+----------+\n",
      "|     James| Bond|      null|\n",
      "|       Ann|Varsa|    Female|\n",
      "|Tom Cruise|  XXX|          |\n",
      "| Tom Brand| null|      Male|\n",
      "+----------+-----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import when\n",
    "df.select(df.fname,df.lname,when(df.gender==\"M\",\"Male\") \\\n",
    "              .when(df.gender==\"F\",\"Female\") \\\n",
    "              .when(df.gender==None ,\"\") \\\n",
    "              .otherwise(df.gender).alias(\"new_gender\") \\\n",
    "    ).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***isin() – Check if value presents in a List.***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+---+\n",
      "|fname|lname| id|\n",
      "+-----+-----+---+\n",
      "|James| Bond|100|\n",
      "|  Ann|Varsa|200|\n",
      "+-----+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "li=[\"100\",\"200\"]\n",
    "df.select(df.fname,df.lname,df.id) \\\n",
    "  .filter(df.id.isin(li)) \\\n",
    "  .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***getField() – To get the value by key from MapType column and by stuct child name from StructType column***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: struct (nullable = true)\n",
      " |    |-- fname: string (nullable = true)\n",
      " |    |-- lname: string (nullable = true)\n",
      " |-- languages: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- properties: map (nullable = true)\n",
      " |    |-- key: string\n",
      " |    |-- value: string (valueContainsNull = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StructType,StructField,StringType,ArrayType,MapType\n",
    "data=[((\"James\",\"Bond\"),[\"Java\",\"C#\"],{'hair':'black','eye':'brown'}),\n",
    "      ((\"Ann\",\"Varsa\"),[\".NET\",\"Python\"],{'hair':'brown','eye':'black'}),\n",
    "      ((\"Tom Cruise\",\"\"),[\"Python\",\"Scala\"],{'hair':'red','eye':'grey'}),\n",
    "      ((\"Tom Brand\",None),[\"Perl\",\"Ruby\"],{'hair':'black','eye':'blue'})]\n",
    "\n",
    "schema = StructType([\n",
    "        StructField('name', StructType([\n",
    "            StructField('fname', StringType(), True),\n",
    "            StructField('lname', StringType(), True)])),\n",
    "        StructField('languages', ArrayType(StringType()),True),\n",
    "        StructField('properties', MapType(StringType(),StringType()),True)\n",
    "     ])\n",
    "df=spark.createDataFrame(data,schema)\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|properties[hair]|\n",
      "+----------------+\n",
      "|           black|\n",
      "|           brown|\n",
      "|             red|\n",
      "|           black|\n",
      "+----------------+\n",
      "\n",
      "+----------+\n",
      "|name.fname|\n",
      "+----------+\n",
      "|     James|\n",
      "|       Ann|\n",
      "|Tom Cruise|\n",
      "| Tom Brand|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#getField from MapType\n",
    "df.select(df.properties.getField(\"hair\")).show()\n",
    "\n",
    "#getField from Struct\n",
    "df.select(df.name.getField(\"fname\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***select() function is used to select single, multiple, column by index, all columns from the list and the nested columns from a DataFrame, PySpark select() is a transformation function hence it returns a new DataFrame with the selected columns***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+-------+-----+\n",
      "|firstname|lastname|country|state|\n",
      "+---------+--------+-------+-----+\n",
      "|James    |Smith   |USA    |CA   |\n",
      "|Michael  |Rose    |USA    |NY   |\n",
      "|Robert   |Williams|USA    |CA   |\n",
      "|Maria    |Jones   |USA    |FL   |\n",
      "+---------+--------+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data = [(\"James\",\"Smith\",\"USA\",\"CA\"),\n",
    "    (\"Michael\",\"Rose\",\"USA\",\"NY\"),\n",
    "    (\"Robert\",\"Williams\",\"USA\",\"CA\"),\n",
    "    (\"Maria\",\"Jones\",\"USA\",\"FL\")\n",
    "  ]\n",
    "columns = [\"firstname\",\"lastname\",\"country\",\"state\"]\n",
    "df = spark.createDataFrame(data = data, schema = columns)\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+\n",
      "|firstname|lastname|\n",
      "+---------+--------+\n",
      "|    James|   Smith|\n",
      "|  Michael|    Rose|\n",
      "|   Robert|Williams|\n",
      "|    Maria|   Jones|\n",
      "+---------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"firstname\",\"lastname\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+\n",
      "|firstname|lastname|\n",
      "+---------+--------+\n",
      "|    James|   Smith|\n",
      "|  Michael|    Rose|\n",
      "|   Robert|Williams|\n",
      "|    Maria|   Jones|\n",
      "+---------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.colRegex(\"`^.*name*`\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***collect() is an action operation that is used to retrieve all the elements of the dataset (from all nodes) to the driver node. We should use the collect() on smaller dataset usually after filter(), group() e.t.c. Retrieving larger datasets results in OutOfMemory error.***"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***withColumn()*** is a transformation function of DataFrame which is used to change the value, convert the datatype of an existing column, create a new column, and many more"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = [('James','','Smith','1991-04-01','M',3000),\n",
    "  ('Michael','Rose','','2000-05-19','M',4000),\n",
    "  ('Robert','','Williams','1978-09-05','M',4000),\n",
    "  ('Maria','Anne','Jones','1967-12-01','F',4000),\n",
    "  ('Jen','Mary','Brown','1980-02-17','F',-1)\n",
    "]\n",
    "\n",
    "columns = [\"firstname\",\"middlename\",\"lastname\",\"dob\",\"gender\",\"salary\"]\n",
    "df = spark.createDataFrame(data=data, schema = columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "|    James|          |   Smith|1991-04-01|     M|  3000|\n",
      "|  Michael|      Rose|        |2000-05-19|     M|  4000|\n",
      "|   Robert|          |Williams|1978-09-05|     M|  4000|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|  4000|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|    -1|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "|    James|          |   Smith|1991-04-01|     M|  3000|\n",
      "|  Michael|      Rose|        |2000-05-19|     M|  4000|\n",
      "|   Robert|          |Williams|1978-09-05|     M|  4000|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|  4000|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|    -1|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Change DataType\n",
    "df.withColumn(\"salary\",col(\"salary\").cast(\"Integer\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "|    James|          |   Smith|1991-04-01|     M|300000|\n",
      "|  Michael|      Rose|        |2000-05-19|     M|400000|\n",
      "|   Robert|          |Williams|1978-09-05|     M|400000|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|400000|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|  -100|\n",
      "+---------+----------+--------+----------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Update The Value of an Existing Column\n",
    "df.withColumn(\"salary\",col(\"salary\")*100).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+------+------------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|CopiedColumn|\n",
      "+---------+----------+--------+----------+------+------+------------+\n",
      "|    James|          |   Smith|1991-04-01|     M|  3000|       -3000|\n",
      "|  Michael|      Rose|        |2000-05-19|     M|  4000|       -4000|\n",
      "|   Robert|          |Williams|1978-09-05|     M|  4000|       -4000|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|  4000|       -4000|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|    -1|           1|\n",
      "+---------+----------+--------+----------+------+------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create a Column from an Existing\n",
    "df.withColumn(\"CopiedColumn\",col(\"salary\")* -1).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+------+-------+\n",
      "|firstname|middlename|lastname|       dob|gender|salary|Country|\n",
      "+---------+----------+--------+----------+------+------+-------+\n",
      "|    James|          |   Smith|1991-04-01|     M|  3000|    USA|\n",
      "|  Michael|      Rose|        |2000-05-19|     M|  4000|    USA|\n",
      "|   Robert|          |Williams|1978-09-05|     M|  4000|    USA|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|  4000|    USA|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|    -1|    USA|\n",
      "+---------+----------+--------+----------+------+------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Add a New Column using withColumn()\n",
    "# create a new column with constant value\n",
    "df.withColumn(\"Country\", lit(\"USA\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+---+------+\n",
      "|firstname|middlename|lastname|dob       |sex|salary|\n",
      "+---------+----------+--------+----------+---+------+\n",
      "|James    |          |Smith   |1991-04-01|M  |3000  |\n",
      "|Michael  |Rose      |        |2000-05-19|M  |4000  |\n",
      "|Robert   |          |Williams|1978-09-05|M  |4000  |\n",
      "|Maria    |Anne      |Jones   |1967-12-01|F  |4000  |\n",
      "|Jen      |Mary      |Brown   |1980-02-17|F  |-1    |\n",
      "+---------+----------+--------+----------+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Rename Column Name\n",
    "df.withColumnRenamed(\"gender\",\"sex\") \\\n",
    "  .show(truncate=False) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------+----------+------+\n",
      "|firstname|middlename|lastname|       dob|gender|\n",
      "+---------+----------+--------+----------+------+\n",
      "|    James|          |   Smith|1991-04-01|     M|\n",
      "|  Michael|      Rose|        |2000-05-19|     M|\n",
      "|   Robert|          |Williams|1978-09-05|     M|\n",
      "|    Maria|      Anne|   Jones|1967-12-01|     F|\n",
      "|      Jen|      Mary|   Brown|1980-02-17|     F|\n",
      "+---------+----------+--------+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# drop Column\n",
    "df.drop(\"salary\") \\\n",
    "  .show() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***filter()*** function is used to filter the rows from RDD/DataFrame based on the given condition or SQL expression, you can also use where() clause instead of the filter() if you are coming from an SQL background, both these functions operate exactly the same."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: struct (nullable = true)\n",
      " |    |-- firstname: string (nullable = true)\n",
      " |    |-- middlename: string (nullable = true)\n",
      " |    |-- lastname: string (nullable = true)\n",
      " |-- languages: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      "\n",
      "+----------------------+------------------+-----+------+\n",
      "|name                  |languages         |state|gender|\n",
      "+----------------------+------------------+-----+------+\n",
      "|{James, , Smith}      |[Java, Scala, C++]|OH   |M     |\n",
      "|{Anna, Rose, }        |[Spark, Java, C++]|NY   |F     |\n",
      "|{Julia, , Williams}   |[CSharp, VB]      |OH   |F     |\n",
      "|{Maria, Anne, Jones}  |[CSharp, VB]      |NY   |M     |\n",
      "|{Jen, Mary, Brown}    |[CSharp, VB]      |NY   |M     |\n",
      "|{Mike, Mary, Williams}|[Python, VB]      |OH   |M     |\n",
      "+----------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StructType,StructField \n",
    "from pyspark.sql.types import StringType, IntegerType, ArrayType\n",
    "data = [\n",
    "    ((\"James\",\"\",\"Smith\"),[\"Java\",\"Scala\",\"C++\"],\"OH\",\"M\"),\n",
    "    ((\"Anna\",\"Rose\",\"\"),[\"Spark\",\"Java\",\"C++\"],\"NY\",\"F\"),\n",
    "    ((\"Julia\",\"\",\"Williams\"),[\"CSharp\",\"VB\"],\"OH\",\"F\"),\n",
    "    ((\"Maria\",\"Anne\",\"Jones\"),[\"CSharp\",\"VB\"],\"NY\",\"M\"),\n",
    "    ((\"Jen\",\"Mary\",\"Brown\"),[\"CSharp\",\"VB\"],\"NY\",\"M\"),\n",
    "    ((\"Mike\",\"Mary\",\"Williams\"),[\"Python\",\"VB\"],\"OH\",\"M\")\n",
    " ]\n",
    "        \n",
    "schema = StructType([\n",
    "     StructField('name', StructType([\n",
    "        StructField('firstname', StringType(), True),\n",
    "        StructField('middlename', StringType(), True),\n",
    "         StructField('lastname', StringType(), True)\n",
    "     ])),\n",
    "     StructField('languages', ArrayType(StringType()), True),\n",
    "     StructField('state', StringType(), True),\n",
    "     StructField('gender', StringType(), True)\n",
    " ])\n",
    "\n",
    "df = spark.createDataFrame(data = data, schema = schema)\n",
    "df.printSchema()\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------+------------------+-----+------+\n",
      "|name                  |languages         |state|gender|\n",
      "+----------------------+------------------+-----+------+\n",
      "|{James, , Smith}      |[Java, Scala, C++]|OH   |M     |\n",
      "|{Julia, , Williams}   |[CSharp, VB]      |OH   |F     |\n",
      "|{Mike, Mary, Williams}|[Python, VB]      |OH   |M     |\n",
      "+----------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Column Condition\n",
    "df.filter(df.state == \"OH\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------------------+-----+------+\n",
      "|               name|         languages|state|gender|\n",
      "+-------------------+------------------+-----+------+\n",
      "|     {Anna, Rose, }|[Spark, Java, C++]|   NY|     F|\n",
      "|{Julia, , Williams}|      [CSharp, VB]|   OH|     F|\n",
      "+-------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SQL Expression\n",
    "df.filter(\"gender != 'M'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------+------------------+-----+------+\n",
      "|name                  |languages         |state|gender|\n",
      "+----------------------+------------------+-----+------+\n",
      "|{James, , Smith}      |[Java, Scala, C++]|OH   |M     |\n",
      "|{Mike, Mary, Williams}|[Python, VB]      |OH   |M     |\n",
      "+----------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Multiple Conditions\n",
    "df.filter( (df.state  == \"OH\") & (df.gender  == \"M\") ) \\\n",
    "    .show(truncate=False) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+-----+------+\n",
      "|                name|         languages|state|gender|\n",
      "+--------------------+------------------+-----+------+\n",
      "|    {James, , Smith}|[Java, Scala, C++]|   OH|     M|\n",
      "| {Julia, , Williams}|      [CSharp, VB]|   OH|     F|\n",
      "|{Mike, Mary, Will...|      [Python, VB]|   OH|     M|\n",
      "+--------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#  List Values\n",
    "li=[\"OH\",\"CA\",\"DE\"]\n",
    "df.filter(df.state.isin(li)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+-----+------+\n",
      "|                name|         languages|state|gender|\n",
      "+--------------------+------------------+-----+------+\n",
      "|      {Anna, Rose, }|[Spark, Java, C++]|   NY|     F|\n",
      "|{Maria, Anne, Jones}|      [CSharp, VB]|   NY|     M|\n",
      "|  {Jen, Mary, Brown}|      [CSharp, VB]|   NY|     M|\n",
      "+--------------------+------------------+-----+------+\n",
      "\n",
      "+--------------------+------------------+-----+------+\n",
      "|                name|         languages|state|gender|\n",
      "+--------------------+------------------+-----+------+\n",
      "|    {James, , Smith}|[Java, Scala, C++]|   OH|     M|\n",
      "| {Julia, , Williams}|      [CSharp, VB]|   OH|     F|\n",
      "|{Mike, Mary, Will...|      [Python, VB]|   OH|     M|\n",
      "+--------------------+------------------+-----+------+\n",
      "\n",
      "+--------------------+------------------+-----+------+\n",
      "|                name|         languages|state|gender|\n",
      "+--------------------+------------------+-----+------+\n",
      "|    {James, , Smith}|[Java, Scala, C++]|   OH|     M|\n",
      "| {Julia, , Williams}|      [CSharp, VB]|   OH|     F|\n",
      "|{Mike, Mary, Will...|      [Python, VB]|   OH|     M|\n",
      "+--------------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Starts With, Ends With, Contains\n",
    "df.filter(df.state.startswith(\"N\")).show()\n",
    "df.filter(df.state.endswith(\"H\")).show()\n",
    "df.filter(df.state.contains(\"H\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "data2 = [(2,\"Michael Rose\"),(3,\"Robert Williams\"),\n",
    "     (4,\"Rames Rose\"),(5,\"Rames rose\")\n",
    "  ]\n",
    "df2 = spark.createDataFrame(data = data2, schema = [\"id\",\"name\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+\n",
      "| id|      name|\n",
      "+---+----------+\n",
      "|  5|Rames rose|\n",
      "+---+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.filter(df2.name.like(\"%rose%\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------------+\n",
      "| id|        name|\n",
      "+---+------------+\n",
      "|  2|Michael Rose|\n",
      "|  4|  Rames Rose|\n",
      "|  5|  Rames rose|\n",
      "+---+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.filter(df2.name.rlike(\"(?i)^*rose$\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+------------------+-----+------+\n",
      "|name            |languages         |state|gender|\n",
      "+----------------+------------------+-----+------+\n",
      "|{James, , Smith}|[Java, Scala, C++]|OH   |M     |\n",
      "|{Anna, Rose, }  |[Spark, Java, C++]|NY   |F     |\n",
      "+----------------+------------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Array column\n",
    "from pyspark.sql.functions import array_contains\n",
    "df.filter(array_contains(df.languages,\"Java\")) \\\n",
    "    .show(truncate=False) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------+------------+-----+------+\n",
      "|name                  |languages   |state|gender|\n",
      "+----------------------+------------+-----+------+\n",
      "|{Julia, , Williams}   |[CSharp, VB]|OH   |F     |\n",
      "|{Mike, Mary, Williams}|[Python, VB]|OH   |M     |\n",
      "+----------------------+------------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Nested Struct columns\n",
    "df.filter(df.name.lastname == \"Williams\") \\\n",
    "    .show(truncate=False) \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***distinct()*** function is used to drop/remove the duplicate rows (all columns) from DataFrame and dropDuplicates() is used to drop rows based on selected (one or multiple) columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = [(\"James\", \"Sales\", 3000), \\\n",
    "    (\"Michael\", \"Sales\", 4600), \\\n",
    "    (\"Robert\", \"Sales\", 4100), \\\n",
    "    (\"Maria\", \"Finance\", 3000), \\\n",
    "    (\"James\", \"Sales\", 3000), \\\n",
    "    (\"Scott\", \"Finance\", 3300), \\\n",
    "    (\"Jen\", \"Finance\", 3900), \\\n",
    "    (\"Jeff\", \"Marketing\", 3000), \\\n",
    "    (\"Kumar\", \"Marketing\", 2000), \\\n",
    "    (\"Saif\", \"Sales\", 4100) \\\n",
    "  ]\n",
    "columns= [\"employee_name\", \"department\", \"salary\"]\n",
    "df = spark.createDataFrame(data = data, schema = columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- employee_name: string (nullable = true)\n",
      " |-- department: string (nullable = true)\n",
      " |-- salary: long (nullable = true)\n",
      "\n",
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|James        |Sales     |3000  |\n",
      "|Michael      |Sales     |4600  |\n",
      "|Robert       |Sales     |4100  |\n",
      "|Maria        |Finance   |3000  |\n",
      "|James        |Sales     |3000  |\n",
      "|Scott        |Finance   |3300  |\n",
      "|Jen          |Finance   |3900  |\n",
      "|Jeff         |Marketing |3000  |\n",
      "|Kumar        |Marketing |2000  |\n",
      "|Saif         |Sales     |4100  |\n",
      "+-------------+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Distinct count: 9\n",
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|Michael      |Sales     |4600  |\n",
      "|James        |Sales     |3000  |\n",
      "|Robert       |Sales     |4100  |\n",
      "|Maria        |Finance   |3000  |\n",
      "|Scott        |Finance   |3300  |\n",
      "|Jen          |Finance   |3900  |\n",
      "|Kumar        |Marketing |2000  |\n",
      "|Jeff         |Marketing |3000  |\n",
      "|Saif         |Sales     |4100  |\n",
      "+-------------+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Distinct\n",
    "distinctDF = df.distinct()\n",
    "print(\"Distinct count: \"+str(distinctDF.count()))\n",
    "distinctDF.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Distinct count: 9\n",
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|Michael      |Sales     |4600  |\n",
      "|James        |Sales     |3000  |\n",
      "|Robert       |Sales     |4100  |\n",
      "|Maria        |Finance   |3000  |\n",
      "|Scott        |Finance   |3300  |\n",
      "|Jen          |Finance   |3900  |\n",
      "|Kumar        |Marketing |2000  |\n",
      "|Jeff         |Marketing |3000  |\n",
      "|Saif         |Sales     |4100  |\n",
      "+-------------+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Drop duplicates\n",
    "df2 = df.dropDuplicates()\n",
    "print(\"Distinct count: \"+str(df2.count()))\n",
    "df2.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Distinct count of department salary : 8\n",
      "+-------------+----------+------+\n",
      "|employee_name|department|salary|\n",
      "+-------------+----------+------+\n",
      "|Maria        |Finance   |3000  |\n",
      "|Scott        |Finance   |3300  |\n",
      "|Jen          |Finance   |3900  |\n",
      "|Kumar        |Marketing |2000  |\n",
      "|Jeff         |Marketing |3000  |\n",
      "|James        |Sales     |3000  |\n",
      "|Robert       |Sales     |4100  |\n",
      "|Michael      |Sales     |4600  |\n",
      "+-------------+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Drop duplicates on selected columns\n",
    "dropDisDF = df.dropDuplicates([\"department\",\"salary\"])\n",
    "print(\"Distinct count of department salary : \"+str(dropDisDF.count()))\n",
    "dropDisDF.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***sort() or orderBy()*** function of PySpark DataFrame to sort DataFrame by ascending or descending order based on single or multiple columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- employee_name: string (nullable = true)\n",
      " |-- department: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- salary: long (nullable = true)\n",
      " |-- age: long (nullable = true)\n",
      " |-- bonus: long (nullable = true)\n",
      "\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Raman        |Finance   |CA   |99000 |40 |24000|\n",
      "|Scott        |Finance   |NY   |83000 |36 |19000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "simpleData = [(\"James\",\"Sales\",\"NY\",90000,34,10000), \\\n",
    "    (\"Michael\",\"Sales\",\"NY\",86000,56,20000), \\\n",
    "    (\"Robert\",\"Sales\",\"CA\",81000,30,23000), \\\n",
    "    (\"Maria\",\"Finance\",\"CA\",90000,24,23000), \\\n",
    "    (\"Raman\",\"Finance\",\"CA\",99000,40,24000), \\\n",
    "    (\"Scott\",\"Finance\",\"NY\",83000,36,19000), \\\n",
    "    (\"Jen\",\"Finance\",\"NY\",79000,53,15000), \\\n",
    "    (\"Jeff\",\"Marketing\",\"CA\",80000,25,18000), \\\n",
    "    (\"Kumar\",\"Marketing\",\"NY\",91000,50,21000) \\\n",
    "  ]\n",
    "columns= [\"employee_name\",\"department\",\"state\",\"salary\",\"age\",\"bonus\"]\n",
    "df = spark.createDataFrame(data = simpleData, schema = columns)\n",
    "df.printSchema()\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Raman        |Finance   |CA   |99000 |40 |24000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "|Scott        |Finance   |NY   |83000 |36 |19000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|Raman        |Finance   |CA   |99000 |40 |24000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "|Scott        |Finance   |NY   |83000 |36 |19000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# sort \n",
    "df.sort(\"department\",\"state\").show(truncate=False)\n",
    "# orderBy\n",
    "df.orderBy(\"department\",\"state\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "|Raman        |Finance   |CA   |99000 |40 |24000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Scott        |Finance   |NY   |83000 |36 |19000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.sort(df.department.desc(),df.state.asc()).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Raman        |Finance   |CA   |99000 |40 |24000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "|Scott        |Finance   |NY   |83000 |36 |19000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Raw SQL\n",
    "df.createOrReplaceTempView(\"EMP\")\n",
    "spark.sql(\"select employee_name,department,state,salary,age,bonus from EMP ORDER BY department asc\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When we perform groupBy() on PySpark Dataframe, it returns GroupedData object which contains below aggregate functions. (count,min,max,avg,mean)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+\n",
      "|department|sum(salary)|\n",
      "+----------+-----------+\n",
      "|Sales     |257000     |\n",
      "|Finance   |351000     |\n",
      "|Marketing |171000     |\n",
      "+----------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import col,sum,avg,max\n",
    "df.groupBy(\"department\").sum(\"salary\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+-----------+----------+\n",
      "|department|state|sum(salary)|sum(bonus)|\n",
      "+----------+-----+-----------+----------+\n",
      "|Sales     |NY   |176000     |30000     |\n",
      "|Sales     |CA   |81000      |23000     |\n",
      "|Finance   |CA   |189000     |47000     |\n",
      "|Finance   |NY   |162000     |34000     |\n",
      "|Marketing |NY   |91000      |21000     |\n",
      "|Marketing |CA   |80000      |18000     |\n",
      "+----------+-----+-----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"department\",\"state\") \\\n",
    "    .sum(\"salary\",\"bonus\") \\\n",
    "    .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+\n",
      "|department|sum_salary|\n",
      "+----------+----------+\n",
      "|Sales     |257000    |\n",
      "|Finance   |351000    |\n",
      "|Marketing |171000    |\n",
      "+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"department\").agg(sum(\"salary\").alias(\"sum_salary\")).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+-----------------+---------+---------+\n",
      "|department|sum_salary|avg_salary       |sum_bonus|max_bonus|\n",
      "+----------+----------+-----------------+---------+---------+\n",
      "|Sales     |257000    |85666.66666666667|53000    |23000    |\n",
      "|Finance   |351000    |87750.0          |81000    |24000    |\n",
      "+----------+----------+-----------------+---------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"department\") \\\n",
    "    .agg(sum(\"salary\").alias(\"sum_salary\"), \\\n",
    "      avg(\"salary\").alias(\"avg_salary\"), \\\n",
    "      sum(\"bonus\").alias(\"sum_bonus\"), \\\n",
    "      max(\"bonus\").alias(\"max_bonus\")) \\\n",
    "    .where(col(\"sum_bonus\") >= 50000) \\\n",
    "    .show(truncate=False)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "PySpark ***Join*** is used to combine two DataFrames and by chaining these you can join multiple DataFrames; it supports all basic join type operations available in traditional SQL like INNER, LEFT OUTER, RIGHT OUTER, LEFT ANTI, LEFT SEMI, CROSS, SELF JOIN. PySpark Joins are wider transformations that involve data shuffling across the network."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- emp_id: long (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- superior_emp_id: long (nullable = true)\n",
      " |-- year_joined: string (nullable = true)\n",
      " |-- emp_dept_id: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- salary: long (nullable = true)\n",
      "\n",
      "+------+--------+---------------+-----------+-----------+------+------+\n",
      "|emp_id|name    |superior_emp_id|year_joined|emp_dept_id|gender|salary|\n",
      "+------+--------+---------------+-----------+-----------+------+------+\n",
      "|1     |Smith   |-1             |2018       |10         |M     |3000  |\n",
      "|2     |Rose    |1              |2010       |20         |M     |4000  |\n",
      "|3     |Williams|1              |2010       |10         |M     |1000  |\n",
      "|4     |Jones   |2              |2005       |10         |F     |2000  |\n",
      "|5     |Brown   |2              |2010       |40         |      |-1    |\n",
      "|6     |Brown   |2              |2010       |50         |      |-1    |\n",
      "+------+--------+---------------+-----------+-----------+------+------+\n",
      "\n",
      "root\n",
      " |-- dept_name: string (nullable = true)\n",
      " |-- dept_id: long (nullable = true)\n",
      "\n",
      "+---------+-------+\n",
      "|dept_name|dept_id|\n",
      "+---------+-------+\n",
      "|Finance  |10     |\n",
      "|Marketing|20     |\n",
      "|Sales    |30     |\n",
      "|IT       |40     |\n",
      "+---------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "emp = [(1,\"Smith\",-1,\"2018\",\"10\",\"M\",3000), \\\n",
    "    (2,\"Rose\",1,\"2010\",\"20\",\"M\",4000), \\\n",
    "    (3,\"Williams\",1,\"2010\",\"10\",\"M\",1000), \\\n",
    "    (4,\"Jones\",2,\"2005\",\"10\",\"F\",2000), \\\n",
    "    (5,\"Brown\",2,\"2010\",\"40\",\"\",-1), \\\n",
    "      (6,\"Brown\",2,\"2010\",\"50\",\"\",-1) \\\n",
    "  ]\n",
    "empColumns = [\"emp_id\",\"name\",\"superior_emp_id\",\"year_joined\", \\\n",
    "       \"emp_dept_id\",\"gender\",\"salary\"]\n",
    "\n",
    "empDF = spark.createDataFrame(data=emp, schema = empColumns)\n",
    "empDF.printSchema()\n",
    "empDF.show(truncate=False)\n",
    "\n",
    "dept = [(\"Finance\",10), \\\n",
    "    (\"Marketing\",20), \\\n",
    "    (\"Sales\",30), \\\n",
    "    (\"IT\",40) \\\n",
    "  ]\n",
    "deptColumns = [\"dept_name\",\"dept_id\"]\n",
    "deptDF = spark.createDataFrame(data=dept, schema = deptColumns)\n",
    "deptDF.printSchema()\n",
    "deptDF.show(truncate=False)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|emp_id|name    |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|1     |Smith   |-1             |2018       |10         |M     |3000  |Finance  |10     |\n",
      "|3     |Williams|1              |2010       |10         |M     |1000  |Finance  |10     |\n",
      "|4     |Jones   |2              |2005       |10         |F     |2000  |Finance  |10     |\n",
      "|2     |Rose    |1              |2010       |20         |M     |4000  |Marketing|20     |\n",
      "|5     |Brown   |2              |2010       |40         |      |-1    |IT       |40     |\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Inner joins two datasets on key columns, where keys don’t match the rows get dropped from both datasets. \n",
    "empDF.join(deptDF,empDF.emp_dept_id ==  deptDF.dept_id,\"inner\") \\\n",
    "     .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|emp_id|name    |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|1     |Smith   |-1             |2018       |10         |M     |3000  |Finance  |10     |\n",
      "|3     |Williams|1              |2010       |10         |M     |1000  |Finance  |10     |\n",
      "|4     |Jones   |2              |2005       |10         |F     |2000  |Finance  |10     |\n",
      "|2     |Rose    |1              |2010       |20         |M     |4000  |Marketing|20     |\n",
      "|null  |null    |null           |null       |null       |null  |null  |Sales    |30     |\n",
      "|5     |Brown   |2              |2010       |40         |      |-1    |IT       |40     |\n",
      "|6     |Brown   |2              |2010       |50         |      |-1    |null     |null   |\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Outer a.k.a full, fullouter join returns all rows from both datasets, where join expression doesn’t match it returns null on respective record columns.\n",
    "empDF.join(deptDF,empDF.emp_dept_id ==  deptDF.dept_id,\"outer\") \\\n",
    "    .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|emp_id|name    |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|1     |Smith   |-1             |2018       |10         |M     |3000  |Finance  |10     |\n",
      "|3     |Williams|1              |2010       |10         |M     |1000  |Finance  |10     |\n",
      "|4     |Jones   |2              |2005       |10         |F     |2000  |Finance  |10     |\n",
      "|2     |Rose    |1              |2010       |20         |M     |4000  |Marketing|20     |\n",
      "|5     |Brown   |2              |2010       |40         |      |-1    |IT       |40     |\n",
      "|6     |Brown   |2              |2010       |50         |      |-1    |null     |null   |\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Left a.k.a Leftouter join returns all rows from the left dataset regardless of match found on the \n",
    "# right dataset when join expression doesn’t match, it assigns null for that record and drops records \n",
    "# from right where match not found\n",
    "empDF.join(deptDF,empDF.emp_dept_id ==  deptDF.dept_id,\"left\")\\\n",
    "    .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|emp_id|name    |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|1     |Smith   |-1             |2018       |10         |M     |3000  |Finance  |10     |\n",
      "|3     |Williams|1              |2010       |10         |M     |1000  |Finance  |10     |\n",
      "|4     |Jones   |2              |2005       |10         |F     |2000  |Finance  |10     |\n",
      "|2     |Rose    |1              |2010       |20         |M     |4000  |Marketing|20     |\n",
      "|null  |null    |null           |null       |null       |null  |null  |Sales    |30     |\n",
      "|5     |Brown   |2              |2010       |40         |      |-1    |IT       |40     |\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Right a.k.a Rightouter join is opposite of left join, here it returns all rows from the right dataset regardless \n",
    "# of math found on the left dataset, when join expression doesn’t match, it assigns null for that record and drops \n",
    "# records from left where match not found. \n",
    "\n",
    "empDF.join(deptDF,empDF.emp_dept_id ==  deptDF.dept_id,\"right\") \\\n",
    "   .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+---------------+-----------+-----------+------+------+\n",
      "|emp_id|name    |superior_emp_id|year_joined|emp_dept_id|gender|salary|\n",
      "+------+--------+---------------+-----------+-----------+------+------+\n",
      "|1     |Smith   |-1             |2018       |10         |M     |3000  |\n",
      "|3     |Williams|1              |2010       |10         |M     |1000  |\n",
      "|4     |Jones   |2              |2005       |10         |F     |2000  |\n",
      "|2     |Rose    |1              |2010       |20         |M     |4000  |\n",
      "|5     |Brown   |2              |2010       |40         |      |-1    |\n",
      "+------+--------+---------------+-----------+-----------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# leftsemi join is similar to inner join difference being leftsemi join returns all columns from the left dataset \n",
    "# and ignores all columns from the right dataset. In other words, this join returns columns from the only left \n",
    "# dataset for the records match in the right dataset on join expression, records not matched on join expression \n",
    "# are ignored from both left and right datasets. \n",
    "empDF.join(deptDF,empDF.emp_dept_id ==  deptDF.dept_id,\"leftsemi\") \\\n",
    "   .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+---------------+-----------+-----------+------+------+\n",
      "|emp_id|name |superior_emp_id|year_joined|emp_dept_id|gender|salary|\n",
      "+------+-----+---------------+-----------+-----------+------+------+\n",
      "|6     |Brown|2              |2010       |50         |      |-1    |\n",
      "+------+-----+---------------+-----------+-----------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# leftanti join does the exact opposite of the leftsemi, leftanti join returns only columns from the left \n",
    "# dataset for non-matched records.\n",
    "empDF.join(deptDF,empDF.emp_dept_id ==  deptDF.dept_id,\"leftanti\") \\\n",
    "   .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|emp_id|name    |superior_emp_id|year_joined|emp_dept_id|gender|salary|dept_name|dept_id|\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "|1     |Smith   |-1             |2018       |10         |M     |3000  |Finance  |10     |\n",
      "|3     |Williams|1              |2010       |10         |M     |1000  |Finance  |10     |\n",
      "|4     |Jones   |2              |2005       |10         |F     |2000  |Finance  |10     |\n",
      "|2     |Rose    |1              |2010       |20         |M     |4000  |Marketing|20     |\n",
      "|5     |Brown   |2              |2010       |40         |      |-1    |IT       |40     |\n",
      "+------+--------+---------------+-----------+-----------+------+------+---------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#SQL\n",
    "empDF.createOrReplaceTempView(\"EMP\")\n",
    "deptDF.createOrReplaceTempView(\"DEPT\")\n",
    "\n",
    "joinDF = spark.sql(\"select * from EMP e, DEPT d where e.emp_dept_id == d.dept_id\") \\\n",
    "  .show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "PySpark ***union() and unionByName()*** transformations are used to merge two or more DataFrame’s of the same schema or structure. The difference between unionByName() function and union() is that this function\n",
    "resolves columns by name (not by position)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- employee_name: string (nullable = true)\n",
      " |-- department: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- salary: long (nullable = true)\n",
      " |-- age: long (nullable = true)\n",
      " |-- bonus: long (nullable = true)\n",
      "\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n",
      "root\n",
      " |-- employee_name: string (nullable = true)\n",
      " |-- department: string (nullable = true)\n",
      " |-- state: string (nullable = true)\n",
      " |-- salary: long (nullable = true)\n",
      " |-- age: long (nullable = true)\n",
      " |-- bonus: long (nullable = true)\n",
      "\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|employee_name|department|state|salary|age|bonus|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "|James        |Sales     |NY   |90000 |34 |10000|\n",
      "|Michael      |Sales     |NY   |86000 |56 |20000|\n",
      "|Robert       |Sales     |CA   |81000 |30 |23000|\n",
      "|Maria        |Finance   |CA   |90000 |24 |23000|\n",
      "|Jen          |Finance   |NY   |79000 |53 |15000|\n",
      "|Kumar        |Marketing |NY   |91000 |50 |21000|\n",
      "|Jeff         |Marketing |CA   |80000 |25 |18000|\n",
      "+-------------+----------+-----+------+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "simpleData = [(\"James\",\"Sales\",\"NY\",90000,34,10000), \\\n",
    "    (\"Michael\",\"Sales\",\"NY\",86000,56,20000), \\\n",
    "    (\"Robert\",\"Sales\",\"CA\",81000,30,23000), \\\n",
    "    (\"Maria\",\"Finance\",\"CA\",90000,24,23000) \\\n",
    "  ]\n",
    "\n",
    "columns= [\"employee_name\",\"department\",\"state\",\"salary\",\"age\",\"bonus\"]\n",
    "df = spark.createDataFrame(data = simpleData, schema = columns)\n",
    "df.printSchema()\n",
    "df.show(truncate=False)\n",
    "\n",
    "simpleData2 = [(\"James\",\"Sales\",\"NY\",90000,34,10000), \\\n",
    "    (\"Maria\",\"Finance\",\"CA\",90000,24,23000), \\\n",
    "    (\"Jen\",\"Finance\",\"NY\",79000,53,15000), \\\n",
    "    (\"Jeff\",\"Marketing\",\"CA\",80000,25,18000), \\\n",
    "    (\"Kumar\",\"Marketing\",\"NY\",91000,50,21000) \\\n",
    "  ]\n",
    "columns2= [\"employee_name\",\"department\",\"state\",\"salary\",\"age\",\"bonus\"]\n",
    "\n",
    "df2 = spark.createDataFrame(data = simpleData2, schema = columns2)\n",
    "\n",
    "df2.printSchema()\n",
    "df2.show(truncate=False)\n",
    "\n",
    "unionDF = df.union(df2)\n",
    "unionDF.show(truncate=False)\n",
    "disDF = df.union(df2).distinct()\n",
    "disDF.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***UDF***’s a.k.a User Defined Functions, If you are coming from SQL background, UDF’s are nothing new to you as most of the traditional RDBMS databases support User Defined Functions, these functions need to register in the database library and use them on SQL as regular functions. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col, udf\n",
    "from pyspark.sql.types import StringType\n",
    "\n",
    "def convertCase(str):\n",
    "    resStr=\"\"\n",
    "    arr = str.split(\" \")\n",
    "    for x in arr:\n",
    "       resStr= resStr + x[0:1].upper() + x[1:len(x)] + \" \"\n",
    "    return resStr "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "convertUDF = udf(lambda z: convertCase(z),StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------------+\n",
      "|Seqno|Name        |\n",
      "+-----+------------+\n",
      "|1    |john jones  |\n",
      "|2    |tracey smith|\n",
      "|3    |amy sanders |\n",
      "+-----+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "columns = [\"Seqno\",\"Name\"]\n",
    "data = [(\"1\", \"john jones\"),\n",
    "    (\"2\", \"tracey smith\"),\n",
    "    (\"3\", \"amy sanders\")]\n",
    "\n",
    "df = spark.createDataFrame(data=data,schema=columns)\n",
    "\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------------+\n",
      "|Seqno|Name         |\n",
      "+-----+-------------+\n",
      "|1    |John Jones   |\n",
      "|2    |Tracey Smith |\n",
      "|3    |Amy Sanders  |\n",
      "+-----+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(col(\"Seqno\"), \\\n",
    "    convertUDF(df[\"Name\"]).alias(\"Name\") ) \\\n",
    ".show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------------+\n",
      "|Seqno|Name         |\n",
      "+-----+-------------+\n",
      "|1    |John Jones   |\n",
      "|2    |Tracey Smith |\n",
      "|3    |Amy Sanders  |\n",
      "+-----+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.udf.register(\"convertUDF\", convertCase,StringType())\n",
    "df.createOrReplaceTempView(\"NAME_TABLE\")\n",
    "spark.sql(\"select Seqno, convertUDF(Name) as Name from NAME_TABLE\") \\\n",
    "     .show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------------+-------------+\n",
      "|Seqno|Name        |Cureated Name|\n",
      "+-----+------------+-------------+\n",
      "|1    |john jones  |JOHN JONES   |\n",
      "|2    |tracey smith|TRACEY SMITH |\n",
      "|3    |amy sanders |AMY SANDERS  |\n",
      "+-----+------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "## best implementation\n",
    "@udf(returnType=StringType()) \n",
    "def upperCase(str):\n",
    "    return str.upper()\n",
    "\n",
    "df.withColumn(\"Cureated Name\", upperCase(col(\"Name\"))) \\\n",
    ".show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**map** is an RDD transformation that is used to apply the transformation function (lambda) on every element of RDD/DataFrame and returns a new RDD:<br>\n",
    "\n",
    "\n",
    "RDD map() transformation is used to apply any complex operations like adding a column, updating a column, transforming the data e.t.c, the output of map transformations would always have the same number of records as input.\n",
    "\n",
    "    Note1: DataFrame doesn’t have map() transformation to use with DataFrame hence you need to DataFrame to RDD first.\n",
    "    Note2: If you have a heavy initialization use PySpark mapPartitions() transformation instead of map(), as with mapPartitions() heavy initialization executes only once for each partition instead of every record."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = [\"Project\",\"Gutenberg’s\",\"Alice’s\",\"Adventures\",\n",
    "\"in\",\"Wonderland\",\"Project\",\"Gutenberg’s\",\"Adventures\",\n",
    "\"in\",\"Wonderland\",\"Project\",\"Gutenberg’s\"]\n",
    "\n",
    "rdd=spark.sparkContext.parallelize(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Project', 1)\n",
      "('Gutenberg’s', 1)\n",
      "('Alice’s', 1)\n",
      "('Adventures', 1)\n",
      "('in', 1)\n",
      "('Wonderland', 1)\n",
      "('Project', 1)\n",
      "('Gutenberg’s', 1)\n",
      "('Adventures', 1)\n",
      "('in', 1)\n",
      "('Wonderland', 1)\n",
      "('Project', 1)\n",
      "('Gutenberg’s', 1)\n"
     ]
    }
   ],
   "source": [
    "rdd2=rdd.map(lambda x: (x,1))\n",
    "for element in rdd2.collect():\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+------+------+\n",
      "|firstname|lastname|gender|salary|\n",
      "+---------+--------+------+------+\n",
      "|    James|   Smith|     M|    30|\n",
      "|     Anna|    Rose|     F|    41|\n",
      "|   Robert|Williams|     M|    62|\n",
      "+---------+--------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data = [('James','Smith','M',30),\n",
    "  ('Anna','Rose','F',41),\n",
    "  ('Robert','Williams','M',62), \n",
    "]\n",
    "\n",
    "columns = [\"firstname\",\"lastname\",\"gender\",\"salary\"]\n",
    "df = spark.createDataFrame(data=data, schema = columns)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+------+----------+\n",
      "|           name|gender|new_salary|\n",
      "+---------------+------+----------+\n",
      "|    James,Smith|     M|        60|\n",
      "|      Anna,Rose|     F|        82|\n",
      "|Robert,Williams|     M|       124|\n",
      "+---------------+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rdd2=df.rdd.map(lambda x: \n",
    "    (x[0]+\",\"+x[1],x[2],x[3]*2)\n",
    "    )  \n",
    "df2=rdd2.toDF([\"name\",\"gender\",\"new_salary\"]   )\n",
    "df2.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+------+----------+\n",
      "|           name|gender|new_salary|\n",
      "+---------------+------+----------+\n",
      "|    James,Smith|     M|        60|\n",
      "|      Anna,Rose|     F|        82|\n",
      "|Robert,Williams|     M|       124|\n",
      "+---------------+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Referring Column Names\n",
    "rdd2=df.rdd.map(lambda x: \n",
    "    (x[\"firstname\"]+\",\"+x[\"lastname\"],x[\"gender\"],x[\"salary\"]*2)\n",
    "    ) \n",
    "rdd2.toDF([\"name\",\"gender\",\"new_salary\"]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "# By Calling function\n",
    "def func1(x):\n",
    "    firstName=x.firstname\n",
    "    lastName=x.lastname\n",
    "    name=firstName+\",\"+lastName\n",
    "    gender=x.gender.lower()\n",
    "    salary=x.salary*2\n",
    "    return (name,gender,salary)\n",
    "\n",
    "rdd2=df.rdd.map(lambda x: func1(x))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**flatMap** is a transformation operation that flattens the RDD/DataFrame (array/map DataFrame columns) after applying the function on every element and returns a new PySpark RDD/DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = [\"Project Gutenberg’s\",\n",
    "        \"Alice’s Adventures in Wonderland\",\n",
    "        \"Project Gutenberg’s\",\n",
    "        \"Adventures in Wonderland\",\n",
    "        \"Project Gutenberg’s\"]\n",
    "rdd=spark.sparkContext.parallelize(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Project\n",
      "Gutenberg’s\n",
      "Alice’s\n",
      "Adventures\n",
      "in\n",
      "Wonderland\n",
      "Project\n",
      "Gutenberg’s\n",
      "Adventures\n",
      "in\n",
      "Wonderland\n",
      "Project\n",
      "Gutenberg’s\n"
     ]
    }
   ],
   "source": [
    "rdd2=rdd.flatMap(lambda x: x.split(\" \"))\n",
    "for element in rdd2.collect():\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- col: string (nullable = true)\n",
      "\n",
      "+---------+------+\n",
      "|     name|   col|\n",
      "+---------+------+\n",
      "|    James|  Java|\n",
      "|    James| Scala|\n",
      "|  Michael| Spark|\n",
      "|  Michael|  Java|\n",
      "|  Michael|  null|\n",
      "|   Robert|CSharp|\n",
      "|   Robert|      |\n",
      "|Jefferson|     1|\n",
      "|Jefferson|     2|\n",
      "+---------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "arrayData = [\n",
    "        ('James',['Java','Scala'],{'hair':'black','eye':'brown'}),\n",
    "        ('Michael',['Spark','Java',None],{'hair':'brown','eye':None}),\n",
    "        ('Robert',['CSharp',''],{'hair':'red','eye':''}),\n",
    "        ('Washington',None,None),\n",
    "        ('Jefferson',['1','2'],{})]\n",
    "df = spark.createDataFrame(data=arrayData, schema = ['name','knownLanguages','properties'])\n",
    "\n",
    "from pyspark.sql.functions import explode\n",
    "df2 = df.select(df.name,explode(df.knownLanguages))\n",
    "df2.printSchema()\n",
    "df2.show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***foreach()*** to Loop Through Rows in DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = [('James','Smith','M',30),('Anna','Rose','F',41),\n",
    "  ('Robert','Williams','M',62), \n",
    "]\n",
    "columns = [\"firstname\",\"lastname\",\"gender\",\"salary\"]\n",
    "df = spark.createDataFrame(data=data, schema = columns)\n",
    "\n",
    "# Foreach example\n",
    "def f(x): print(x)\n",
    "df.foreach(f)\n",
    "\n",
    "# Another example\n",
    "df.foreach(lambda x: \n",
    "    print(\"Data ==>\"+x[\"firstname\"]+\",\"+x[\"lastname\"]+\",\"+x[\"gender\"]+\",\"+str(x[\"salary\"]*2))\n",
    "    ) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "sample from dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(id=13), Row(id=21), Row(id=35), Row(id=66), Row(id=69), Row(id=70), Row(id=87)]\n",
      "[Row(id=19), Row(id=45), Row(id=74), Row(id=75), Row(id=84), Row(id=88)]\n"
     ]
    }
   ],
   "source": [
    "df=spark.range(100)\n",
    "\n",
    "# without replacement\n",
    "print(df.sample(0.06).collect())\n",
    "\n",
    "# with replacement\n",
    "print(df.sample(True,0.06,123).collect()) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***fillna()*** or ***DataFrameNaFunctions.fill()*** is used to replace NULL/None values on all or selected multiple DataFrame columns with either zero(0), empty string, space, or any constant literal values. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['firstname', 'lastname', 'gender', 'salary']"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+------+------+\n",
      "|firstname|lastname|gender|salary|\n",
      "+---------+--------+------+------+\n",
      "|    James|   Smith|     M|    30|\n",
      "|     Anna|    Rose|     F|    41|\n",
      "|   Robert|Williams|     M|    62|\n",
      "+---------+--------+------+------+\n",
      "\n",
      "+---------+--------+------+------+\n",
      "|firstname|lastname|gender|salary|\n",
      "+---------+--------+------+------+\n",
      "|    James|   Smith|     M|    30|\n",
      "|     Anna|    Rose|     F|    41|\n",
      "|   Robert|Williams|     M|    62|\n",
      "+---------+--------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Replace 0 for null for all integer columns\n",
    "df.na.fill(value=0).show()\n",
    "\n",
    "#Replace 0 for null on only population column \n",
    "df.na.fill(value=0,subset=[\"salary\"]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+------+------+\n",
      "|firstname|lastname|gender|salary|\n",
      "+---------+--------+------+------+\n",
      "|    James|   Smith|     M|    30|\n",
      "|     Anna|    Rose|     F|    41|\n",
      "|   Robert|Williams|     M|    62|\n",
      "+---------+--------+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "\n",
    "df.na.fill({\"firstname\": \"unknown\", \"salary\": 10}) \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***pivot()*** function is used to rotate/transpose the data from one column into multiple Dataframe columns and back using unpivot(). Pivot() It is an aggregation where one of the grouping columns values is transposed into individual columns with distinct data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Product: string (nullable = true)\n",
      " |-- Amount: long (nullable = true)\n",
      " |-- Country: string (nullable = true)\n",
      "\n",
      "+-------+------+-------+\n",
      "|Product|Amount|Country|\n",
      "+-------+------+-------+\n",
      "|Banana |1000  |USA    |\n",
      "|Carrots|1500  |USA    |\n",
      "|Beans  |1600  |USA    |\n",
      "|Orange |2000  |USA    |\n",
      "|Orange |2000  |USA    |\n",
      "|Banana |400   |China  |\n",
      "|Carrots|1200  |China  |\n",
      "|Beans  |1500  |China  |\n",
      "|Orange |4000  |China  |\n",
      "|Banana |2000  |Canada |\n",
      "|Carrots|2000  |Canada |\n",
      "|Beans  |2000  |Mexico |\n",
      "+-------+------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data = [(\"Banana\",1000,\"USA\"), (\"Carrots\",1500,\"USA\"), (\"Beans\",1600,\"USA\"), \\\n",
    "      (\"Orange\",2000,\"USA\"),(\"Orange\",2000,\"USA\"),(\"Banana\",400,\"China\"), \\\n",
    "      (\"Carrots\",1200,\"China\"),(\"Beans\",1500,\"China\"),(\"Orange\",4000,\"China\"), \\\n",
    "      (\"Banana\",2000,\"Canada\"),(\"Carrots\",2000,\"Canada\"),(\"Beans\",2000,\"Mexico\")]\n",
    "\n",
    "columns= [\"Product\",\"Amount\",\"Country\"]\n",
    "df = spark.createDataFrame(data = data, schema = columns)\n",
    "df.printSchema()\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Product: string (nullable = true)\n",
      " |-- Canada: long (nullable = true)\n",
      " |-- China: long (nullable = true)\n",
      " |-- Mexico: long (nullable = true)\n",
      " |-- USA: long (nullable = true)\n",
      "\n",
      "+-------+------+-----+------+----+\n",
      "|Product|Canada|China|Mexico|USA |\n",
      "+-------+------+-----+------+----+\n",
      "|Orange |null  |4000 |null  |4000|\n",
      "|Beans  |null  |1500 |2000  |1600|\n",
      "|Banana |2000  |400  |null  |1000|\n",
      "|Carrots|2000  |1200 |null  |1500|\n",
      "+-------+------+-----+------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pivotDF = df.groupBy(\"Product\").pivot(\"Country\").sum(\"Amount\")\n",
    "pivotDF.printSchema()\n",
    "pivotDF.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-----+\n",
      "|Product|Country|Total|\n",
      "+-------+-------+-----+\n",
      "|Orange |China  |4000 |\n",
      "|Beans  |China  |1500 |\n",
      "|Beans  |Mexico |2000 |\n",
      "|Banana |Canada |2000 |\n",
      "|Banana |China  |400  |\n",
      "|Carrots|Canada |2000 |\n",
      "|Carrots|China  |1200 |\n",
      "+-------+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import expr\n",
    "unpivotExpr = \"stack(3, 'Canada', Canada, 'China', China, 'Mexico', Mexico) as (Country,Total)\"\n",
    "unPivotDF = pivotDF.select(\"Product\", expr(unpivotExpr)) \\\n",
    "    .where(\"Total is not null\")\n",
    "unPivotDF.show(truncate=False)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "***partitionBy()*** is a function of pyspark.sql.DataFrameWriter class which is used to partition the large dataset (DataFrame) into smaller files based on one or multiple columns while writing to disk,"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "___Partition in memory___: You can partition or repartition the DataFrame by calling repartition() or coalesce() transformations.\n",
    "\n",
    "___Partition on disk___: While writing the PySpark DataFrame back to disk, you can choose how to partition the data based on columns using partitionBy() of pyspark.sql.DataFrameWriter."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "PySpark ___ArrayType___ is a collection data type that extends the DataType class which is a superclass of all types in PySpark. All elements of ArrayType should have the same type of elements."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- languagesAtSchool: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- languagesAtWork: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- currentState: string (nullable = true)\n",
      " |-- previousState: string (nullable = true)\n",
      "\n",
      "+----------------+------------------+---------------+------------+-------------+\n",
      "|            name| languagesAtSchool|languagesAtWork|currentState|previousState|\n",
      "+----------------+------------------+---------------+------------+-------------+\n",
      "|    James,,Smith|[Java, Scala, C++]|  [Spark, Java]|          OH|           CA|\n",
      "|   Michael,Rose,|[Spark, Java, C++]|  [Spark, Java]|          NY|           NJ|\n",
      "|Robert,,Williams|      [CSharp, VB]|[Spark, Python]|          UT|           NV|\n",
      "+----------------+------------------+---------------+------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data = [\n",
    " (\"James,,Smith\",[\"Java\",\"Scala\",\"C++\"],[\"Spark\",\"Java\"],\"OH\",\"CA\"),\n",
    " (\"Michael,Rose,\",[\"Spark\",\"Java\",\"C++\"],[\"Spark\",\"Java\"],\"NY\",\"NJ\"),\n",
    " (\"Robert,,Williams\",[\"CSharp\",\"VB\"],[\"Spark\",\"Python\"],\"UT\",\"NV\")\n",
    "]\n",
    "\n",
    "from pyspark.sql.types import StringType, ArrayType,StructType,StructField\n",
    "schema = StructType([ \n",
    "    StructField(\"name\",StringType(),True), \n",
    "    StructField(\"languagesAtSchool\",ArrayType(StringType()),True), \n",
    "    StructField(\"languagesAtWork\",ArrayType(StringType()),True), \n",
    "    StructField(\"currentState\", StringType(), True), \n",
    "    StructField(\"previousState\", StringType(), True)\n",
    "  ])\n",
    "\n",
    "df = spark.createDataFrame(data=data,schema=schema)\n",
    "df.printSchema()\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use ___explode()___ function to create a new row for each element in the given array column. There are various PySpark SQL explode functions available to work with Array columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+------+\n",
      "|            name|   col|\n",
      "+----------------+------+\n",
      "|    James,,Smith|  Java|\n",
      "|    James,,Smith| Scala|\n",
      "|    James,,Smith|   C++|\n",
      "|   Michael,Rose,| Spark|\n",
      "|   Michael,Rose,|  Java|\n",
      "|   Michael,Rose,|   C++|\n",
      "|Robert,,Williams|CSharp|\n",
      "|Robert,,Williams|    VB|\n",
      "+----------------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import explode\n",
    "df.select(df.name,explode(df.languagesAtSchool)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "___split()___ sql function returns an array type after splitting the string column by delimiter. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|         nameAsArray|\n",
      "+--------------------+\n",
      "|    [James, , Smith]|\n",
      "|   [Michael, Rose, ]|\n",
      "|[Robert, , Williams]|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import split\n",
    "df.select(split(df.name,\",\").alias(\"nameAsArray\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use ___array()___ function to create a new array column by merging the data from multiple columns. All input columns must have the same data type."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------+\n",
      "|            name|  States|\n",
      "+----------------+--------+\n",
      "|    James,,Smith|[OH, CA]|\n",
      "|   Michael,Rose,|[NY, NJ]|\n",
      "|Robert,,Williams|[UT, NV]|\n",
      "+----------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import array\n",
    "df.select(df.name,array(df.currentState,df.previousState).alias(\"States\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "___array_contains()___ sql function is used to check if array column contains a value. Returns null if the array is null, true if the array contains the value, and false otherwise."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------+\n",
      "|            name|array_contains|\n",
      "+----------------+--------------+\n",
      "|    James,,Smith|          true|\n",
      "|   Michael,Rose,|          true|\n",
      "|Robert,,Williams|         false|\n",
      "+----------------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import array_contains\n",
    "df.select(df.name,array_contains(df.languagesAtSchool,\"Java\")\n",
    "    .alias(\"array_contains\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "PySpark ___MapType___ (also called map type) is a data type to represent Python Dictionary (dict) to store key-value pair, a MapType object comprises three fields, keyType (a DataType), valueType (a DataType) and valueContainsNull (a BooleanType)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructField, StructType, StringType, MapType\n",
    "schema = StructType([\n",
    "    StructField('name', StringType(), True),\n",
    "    StructField('properties', MapType(StringType(),StringType()),True)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- properties: map (nullable = true)\n",
      " |    |-- key: string\n",
      " |    |-- value: string (valueContainsNull = true)\n",
      "\n",
      "+----------+-----------------------------+\n",
      "|name      |properties                   |\n",
      "+----------+-----------------------------+\n",
      "|James     |{eye -> brown, hair -> black}|\n",
      "|Michael   |{eye -> null, hair -> brown} |\n",
      "|Robert    |{eye -> black, hair -> red}  |\n",
      "|Washington|{eye -> grey, hair -> grey}  |\n",
      "|Jefferson |{eye -> , hair -> brown}     |\n",
      "+----------+-----------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataDictionary = [\n",
    "        ('James',{'hair':'black','eye':'brown'}),\n",
    "        ('Michael',{'hair':'brown','eye':None}),\n",
    "        ('Robert',{'hair':'red','eye':'black'}),\n",
    "        ('Washington',{'hair':'grey','eye':'grey'}),\n",
    "        ('Jefferson',{'hair':'brown','eye':''})\n",
    "        ]\n",
    "df = spark.createDataFrame(data=dataDictionary, schema = schema)\n",
    "df.printSchema()\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- hair: string (nullable = true)\n",
      " |-- eye: string (nullable = true)\n",
      "\n",
      "+----------+-----+-----+\n",
      "|      name| hair|  eye|\n",
      "+----------+-----+-----+\n",
      "|     James|black|brown|\n",
      "|   Michael|brown| null|\n",
      "|    Robert|  red|black|\n",
      "|Washington| grey| grey|\n",
      "| Jefferson|brown|     |\n",
      "+----------+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Access PySpark MapType Elements\n",
    "df3=df.rdd.map(lambda x: \\\n",
    "    (x.name,x.properties[\"hair\"],x.properties[\"eye\"])) \\\n",
    "    .toDF([\"name\",\"hair\",\"eye\"])\n",
    "df3.printSchema()\n",
    "df3.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----+-----+\n",
      "|      name| hair|  eye|\n",
      "+----------+-----+-----+\n",
      "|     James|black|brown|\n",
      "|   Michael|brown| null|\n",
      "|    Robert|  red|black|\n",
      "|Washington| grey| grey|\n",
      "| Jefferson|brown|     |\n",
      "+----------+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.withColumn(\"hair\",df.properties[\"hair\"]) \\\n",
    "  .withColumn(\"eye\",df.properties[\"eye\"]) \\\n",
    "  .drop(\"properties\") \\\n",
    "  .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----+-----+\n",
      "|      name| key|value|\n",
      "+----------+----+-----+\n",
      "|     James| eye|brown|\n",
      "|     James|hair|black|\n",
      "|   Michael| eye| null|\n",
      "|   Michael|hair|brown|\n",
      "|    Robert| eye|black|\n",
      "|    Robert|hair|  red|\n",
      "|Washington| eye| grey|\n",
      "|Washington|hair| grey|\n",
      "| Jefferson| eye|     |\n",
      "| Jefferson|hair|brown|\n",
      "+----------+----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.name,explode(df.properties)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+\n",
      "|      name|map_keys(properties)|\n",
      "+----------+--------------------+\n",
      "|     James|         [eye, hair]|\n",
      "|   Michael|         [eye, hair]|\n",
      "|    Robert|         [eye, hair]|\n",
      "|Washington|         [eye, hair]|\n",
      "| Jefferson|         [eye, hair]|\n",
      "+----------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# map KEYS\n",
    "from pyspark.sql.functions import map_keys\n",
    "df.select(df.name,map_keys(df.properties)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------------------+\n",
      "|      name|map_values(properties)|\n",
      "+----------+----------------------+\n",
      "|     James|        [brown, black]|\n",
      "|   Michael|         [null, brown]|\n",
      "|    Robert|          [black, red]|\n",
      "|Washington|          [grey, grey]|\n",
      "| Jefferson|             [, brown]|\n",
      "+----------+----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# map Values\n",
    "from pyspark.sql.functions import map_values\n",
    "df.select(df.name,map_values(df.properties)).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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}