04 Data Filtering.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark training for data engineers\n",
    "## 04. Data Filtering"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Goal\n",
    "\n",
    "Practise with filtering on the two RDDs created in the earlier notebooks. Both RDDs are converted to dataframes and some simple filtering examples are shown."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Highlights\n",
    "\n",
    "* `sqlContext.createDataFrame()` creates a Spark dataframe from a RDD.\n",
    "* If the schema cannot be inferred from the RDD, a schema has to be supplied when converting the RDD to a dataframe.\n",
    "* Using `dataframe.select()` and `dataframe.where()` data can be selected.\n",
    "* By using `dataframe.groupBy()` data can be sorted (and counted, etc)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Implementation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext\n",
    "config = SparkConf().setMaster('local')\n",
    "spark = SparkContext.getOrCreate(conf=config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### XML"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "xmlrdd = spark.pickleFile('xml-pickle-03/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'text': 'One', 'info': 'testfile'}, {'text': 'Two', 'info': 'testfile'}]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "xmlrdd.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us create a SQLContext to get more flexibility in our Spark environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can convert the RDD to a dataframe which makes it easier to work with the data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/spark/python/pyspark/sql/session.py:360: UserWarning: Using RDD of dict to inferSchema is deprecated. Use pyspark.sql.Row instead\n",
      "  warnings.warn(\"Using RDD of dict to inferSchema is deprecated. \"\n"
     ]
    }
   ],
   "source": [
    "xmldf = sqlContext.createDataFrame(xmlrdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----+\n",
      "|    info|text|\n",
      "+--------+----+\n",
      "|testfile| One|\n",
      "|testfile| Two|\n",
      "+--------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "xmldf.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From the above print we can observe that the columns are correct, both columns (info and text) are inferred from the RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----+\n",
      "|    info|text|\n",
      "+--------+----+\n",
      "|testfile| One|\n",
      "+--------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "xmldf.where(xmldf['text'] == 'One').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----+\n",
      "|    info|text|\n",
      "+--------+----+\n",
      "|testfile| Two|\n",
      "+--------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "xmldf.where(xmldf['text'] == 'Two').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### CSV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "csvrdd = spark.pickleFile('csv-pickle-03/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['john,doe,male,32',\n",
       " 'jake,doe,male,16',\n",
       " 'jane,doe,female,31',\n",
       " 'janet,doe,female,13']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csvrdd.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First we need to create proper Rows from each CSV line by using a mapping."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[<Row(john, doe, male, 32)>,\n",
       " <Row(jake, doe, male, 16)>,\n",
       " <Row(jane, doe, female, 31)>,\n",
       " <Row(janet, doe, female, 13)>]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import Row\n",
    "\n",
    "def processCSV(row):\n",
    "    # Split the row into a list\n",
    "    row = row.split(',')\n",
    "    # Return the four fields\n",
    "    return Row(row[0], row[1], row[2], row[3])\n",
    "\n",
    "csvrdd = csvrdd.map(lambda row: processCSV(row))\n",
    "csvrdd.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We define the schema of the data so the `createDataFrame` does not have to infer the schema."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructField, StructType, StringType, IntegerType\n",
    "schema = StructType([\n",
    "            StructField(\"first_name\", StringType(), True),\n",
    "            StructField(\"last_name\", StringType(), True),\n",
    "            StructField(\"gender\", StringType(), True),\n",
    "            StructField(\"age\", StringType(), True)\n",
    "        ])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "csvdf = sqlContext.createDataFrame(csvrdd, schema=schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+\n",
      "|first_name|last_name|gender|age|\n",
      "+----------+---------+------+---+\n",
      "|      john|      doe|  male| 32|\n",
      "|      jake|      doe|  male| 16|\n",
      "|      jane|      doe|female| 31|\n",
      "|     janet|      doe|female| 13|\n",
      "+----------+---------+------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csvdf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row((age > 30)=True),\n",
       " Row((age > 30)=False),\n",
       " Row((age > 30)=True),\n",
       " Row((age > 30)=False)]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csvdf.select(csvdf.age > 30).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|(age > 30)|\n",
      "+----------+\n",
      "|      true|\n",
      "|     false|\n",
      "|      true|\n",
      "|     false|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csvdf.select(csvdf.age > 30).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(first_name='john', (age > 30)=True),\n",
       " Row(first_name='jake', (age > 30)=False),\n",
       " Row(first_name='jane', (age > 30)=True),\n",
       " Row(first_name='janet', (age > 30)=False)]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csvdf.select('first_name', csvdf.age > 30).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------------------------+\n",
      "|first_name|CASE WHEN (age > 30) THEN 1 ELSE 0 END|\n",
      "+----------+--------------------------------------+\n",
      "|      john|                                     1|\n",
      "|      jake|                                     0|\n",
      "|      jane|                                     1|\n",
      "|     janet|                                     0|\n",
      "+----------+--------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import functions as psf\n",
    "csvdf.select('first_name', psf.when(csvdf.age > 30, 1).otherwise(0)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+\n",
      "|first_name|last_name|gender|age|\n",
      "+----------+---------+------+---+\n",
      "|      jane|      doe|female| 31|\n",
      "|     janet|      doe|female| 13|\n",
      "+----------+---------+------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csvdf[csvdf.first_name.isin(\"jane\",\"janet\")].show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+\n",
      "|first_name|last_name|gender|age|\n",
      "+----------+---------+------+---+\n",
      "|      jake|      doe|  male| 16|\n",
      "|      jane|      doe|female| 31|\n",
      "|     janet|      doe|female| 13|\n",
      "+----------+---------+------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csvdf[csvdf.first_name.startswith(\"ja\")].show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count the different genders:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|gender|count|\n",
      "+------+-----+\n",
      "|female|    2|\n",
      "|  male|    2|\n",
      "+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csvdf.groupBy('gender').count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sort by age:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+\n",
      "|first_name|last_name|gender|age|\n",
      "+----------+---------+------+---+\n",
      "|     janet|      doe|female| 13|\n",
      "|      jake|      doe|  male| 16|\n",
      "|      jane|      doe|female| 31|\n",
      "|      john|      doe|  male| 32|\n",
      "+----------+---------+------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csvdf.sort('age').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Save the dataframe to Parquet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "csvdf.write.parquet('notebook-04-parquet')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "part-00000-9825b615-2bf6-4cdc-862a-3a875977eaf4-c000.snappy.parquet  _SUCCESS\n"
     ]
    }
   ],
   "source": [
    "ls notebook-04-parquet/"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}