05 Data Enriching.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark training for data engineers\n",
    "## 05. Data Enriching"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Goal\n",
    "\n",
    "Adding more value to the data by \n",
    "* Adding new columns\n",
    "* Using lambda functions\n",
    "* Using user defined functions\n",
    "\n",
    "### Highlights\n",
    "* `df.withColumn('new_col', Function())` a new column is added to the DataFrame\n",
    "* `len_fun = udf(lambda z: len(z), IntegerType())` is a User Defined Function that returns the length of the input as integer\n",
    "* `df = df.withColumn('length_col', len_fun('text_col'))` will add a column `length_col` with the length of the item in `text_col`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Implementation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "config = SparkConf().setMaster('local')\n",
    "spark = SparkContext.getOrCreate(conf=config)\n",
    "sqlContext = SQLContext(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = sqlContext.read.parquet('notebook-04-parquet/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "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": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('double_age', df.age*2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+----------+\n",
      "|first_name|last_name|gender|age|double_age|\n",
      "+----------+---------+------+---+----------+\n",
      "|      john|      doe|  male| 32|      64.0|\n",
      "|      jake|      doe|  male| 16|      32.0|\n",
      "|      jane|      doe|female| 31|      62.0|\n",
      "|     janet|      doe|female| 13|      26.0|\n",
      "+----------+---------+------+---+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define a user defined function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import udf\n",
    "\n",
    "@udf('integer')\n",
    "def calc_name_length(name):\n",
    "    return len(name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('name_length', calc_name_length(df.first_name))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+----------+-----------+\n",
      "|first_name|last_name|gender|age|double_age|name_length|\n",
      "+----------+---------+------+---+----------+-----------+\n",
      "|      john|      doe|  male| 32|      64.0|          4|\n",
      "|      jake|      doe|  male| 16|      32.0|          4|\n",
      "|      jane|      doe|female| 31|      62.0|          4|\n",
      "|     janet|      doe|female| 13|      26.0|          5|\n",
      "+----------+---------+------+---+----------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define a lambda function with one input parameter:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import IntegerType\n",
    "len_udf_int = udf(lambda z: len(z), IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('last_name_length', len_udf_int('last_name'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+----------+-----------+----------------+\n",
      "|first_name|last_name|gender|age|double_age|name_length|last_name_lenght|\n",
      "+----------+---------+------+---+----------+-----------+----------------+\n",
      "|      john|      doe|  male| 32|      64.0|          4|               3|\n",
      "|      jake|      doe|  male| 16|      32.0|          4|               3|\n",
      "|      jane|      doe|female| 31|      62.0|          4|               3|\n",
      "|     janet|      doe|female| 13|      26.0|          5|               3|\n",
      "+----------+---------+------+---+----------+-----------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define a lambda function with two input parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "len_udf_two_int = udf(lambda z,y: len(z)+len(y), IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('full_name_length', len_udf_two_int('first_name', 'last_name'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+----------+-----------+----------------+----------------+\n",
      "|first_name|last_name|gender|age|double_age|name_length|last_name_lenght|full_name_length|\n",
      "+----------+---------+------+---+----------+-----------+----------------+----------------+\n",
      "|      john|      doe|  male| 32|      64.0|          4|               3|               7|\n",
      "|      jake|      doe|  male| 16|      32.0|          4|               3|               7|\n",
      "|      jane|      doe|female| 31|      62.0|          4|               3|               7|\n",
      "|     janet|      doe|female| 13|      26.0|          5|               3|               8|\n",
      "+----------+---------+------+---+----------+-----------+----------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Remove a column from the dataframe:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+---------+------+---+-----------+----------------+----------------+\n",
      "|first_name|last_name|gender|age|name_length|last_name_lenght|full_name_length|\n",
      "+----------+---------+------+---+-----------+----------------+----------------+\n",
      "|      john|      doe|  male| 32|          4|               3|               7|\n",
      "|      jake|      doe|  male| 16|          4|               3|               7|\n",
      "|      jane|      doe|female| 31|          4|               3|               7|\n",
      "|     janet|      doe|female| 13|          5|               3|               8|\n",
      "+----------+---------+------+---+-----------+----------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = df.drop('double_age')\n",
    "df.show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}