02_basic_transformations_1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "9913130b-9ed3-4a87-a01d-a13cae0e3204",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://b96cbc5ef428:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.3.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Basic Transformation - I</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7ff4604a6f80>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Spark Session\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = (\n",
    "    SparkSession\n",
    "    .builder\n",
    "    .appName(\"Basic Transformation - I\")\n",
    "    .master(\"local[*]\")\n",
    "    .getOrCreate()\n",
    ")\n",
    "\n",
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "7f6357c0-ccf8-43de-8f7d-69d5a9314fef",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Emp Data & Schema\n",
    "\n",
    "emp_data = [\n",
    "    [\"001\",\"101\",\"John Doe\",\"30\",\"Male\",\"50000\",\"2015-01-01\"],\n",
    "    [\"002\",\"101\",\"Jane Smith\",\"25\",\"Female\",\"45000\",\"2016-02-15\"],\n",
    "    [\"003\",\"102\",\"Bob Brown\",\"35\",\"Male\",\"55000\",\"2014-05-01\"],\n",
    "    [\"004\",\"102\",\"Alice Lee\",\"28\",\"Female\",\"48000\",\"2017-09-30\"],\n",
    "    [\"005\",\"103\",\"Jack Chan\",\"40\",\"Male\",\"60000\",\"2013-04-01\"],\n",
    "    [\"006\",\"103\",\"Jill Wong\",\"32\",\"Female\",\"52000\",\"2018-07-01\"],\n",
    "    [\"007\",\"101\",\"James Johnson\",\"42\",\"Male\",\"70000\",\"2012-03-15\"],\n",
    "    [\"008\",\"102\",\"Kate Kim\",\"29\",\"Female\",\"51000\",\"2019-10-01\"],\n",
    "    [\"009\",\"103\",\"Tom Tan\",\"33\",\"Male\",\"58000\",\"2016-06-01\"],\n",
    "    [\"010\",\"104\",\"Lisa Lee\",\"27\",\"Female\",\"47000\",\"2018-08-01\"],\n",
    "    [\"011\",\"104\",\"David Park\",\"38\",\"Male\",\"65000\",\"2015-11-01\"],\n",
    "    [\"012\",\"105\",\"Susan Chen\",\"31\",\"Female\",\"54000\",\"2017-02-15\"],\n",
    "    [\"013\",\"106\",\"Brian Kim\",\"45\",\"Male\",\"75000\",\"2011-07-01\"],\n",
    "    [\"014\",\"107\",\"Emily Lee\",\"26\",\"Female\",\"46000\",\"2019-01-01\"],\n",
    "    [\"015\",\"106\",\"Michael Lee\",\"37\",\"Male\",\"63000\",\"2014-09-30\"],\n",
    "    [\"016\",\"107\",\"Kelly Zhang\",\"30\",\"Female\",\"49000\",\"2018-04-01\"],\n",
    "    [\"017\",\"105\",\"George Wang\",\"34\",\"Male\",\"57000\",\"2016-03-15\"],\n",
    "    [\"018\",\"104\",\"Nancy Liu\",\"29\",\"Female\",\"50000\",\"2017-06-01\"],\n",
    "    [\"019\",\"103\",\"Steven Chen\",\"36\",\"Male\",\"62000\",\"2015-08-01\"],\n",
    "    [\"020\",\"102\",\"Grace Kim\",\"32\",\"Female\",\"53000\",\"2018-11-01\"]\n",
    "]\n",
    "\n",
    "emp_schema = \"employee_id string, department_id string, name string, age string, gender string, salary string, hire_date string\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "9bdc19a5-b213-498e-8e1f-63650523c441",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create emp DataFrame\n",
    "\n",
    "emp = spark.createDataFrame(data=emp_data, schema=emp_schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "9d81001d-2633-439f-9153-9b1ef27e64a8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------------+-------------+---+------+------+----------+\n",
      "|employee_id|department_id|         name|age|gender|salary| hire_date|\n",
      "+-----------+-------------+-------------+---+------+------+----------+\n",
      "|        001|          101|     John Doe| 30|  Male| 50000|2015-01-01|\n",
      "|        002|          101|   Jane Smith| 25|Female| 45000|2016-02-15|\n",
      "|        003|          102|    Bob Brown| 35|  Male| 55000|2014-05-01|\n",
      "|        004|          102|    Alice Lee| 28|Female| 48000|2017-09-30|\n",
      "|        005|          103|    Jack Chan| 40|  Male| 60000|2013-04-01|\n",
      "|        006|          103|    Jill Wong| 32|Female| 52000|2018-07-01|\n",
      "|        007|          101|James Johnson| 42|  Male| 70000|2012-03-15|\n",
      "|        008|          102|     Kate Kim| 29|Female| 51000|2019-10-01|\n",
      "|        009|          103|      Tom Tan| 33|  Male| 58000|2016-06-01|\n",
      "|        010|          104|     Lisa Lee| 27|Female| 47000|2018-08-01|\n",
      "|        011|          104|   David Park| 38|  Male| 65000|2015-11-01|\n",
      "|        012|          105|   Susan Chen| 31|Female| 54000|2017-02-15|\n",
      "|        013|          106|    Brian Kim| 45|  Male| 75000|2011-07-01|\n",
      "|        014|          107|    Emily Lee| 26|Female| 46000|2019-01-01|\n",
      "|        015|          106|  Michael Lee| 37|  Male| 63000|2014-09-30|\n",
      "|        016|          107|  Kelly Zhang| 30|Female| 49000|2018-04-01|\n",
      "|        017|          105|  George Wang| 34|  Male| 57000|2016-03-15|\n",
      "|        018|          104|    Nancy Liu| 29|Female| 50000|2017-06-01|\n",
      "|        019|          103|  Steven Chen| 36|  Male| 62000|2015-08-01|\n",
      "|        020|          102|    Grace Kim| 32|Female| 53000|2018-11-01|\n",
      "+-----------+-------------+-------------+---+------+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Show emp dataframe (ACTION)\n",
    "\n",
    "emp.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "40e8d717-c449-42f4-a3e3-6bff56077b6b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType([StructField('employee_id', StringType(), True), StructField('department_id', StringType(), True), StructField('name', StringType(), True), StructField('age', StringType(), True), StructField('gender', StringType(), True), StructField('salary', StringType(), True), StructField('hire_date', StringType(), True)])"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Schema for emp\n",
    "\n",
    "emp.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "ae92a9a3-dd82-41b0-ae7d-bbb341371bb2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Small Example for Schema\n",
    "from pyspark.sql.types import StructType, StructField, StringType, IntegerType\n",
    "schema_string = \"name string, age int\"\n",
    "\n",
    "schema_spark =  StructType([\n",
    "    StructField(\"name\", StringType(), True),\n",
    "    StructField(\"age\", IntegerType(), True)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "f5a249be-bfab-4bde-ae61-eb6782660b49",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Column<'salary'>"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Columns and expression\n",
    "from pyspark.sql.functions import col, expr\n",
    "\n",
    "emp[\"salary\"]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "ae885c86-5d7a-4744-8ea7-d16849e68eea",
   "metadata": {},
   "outputs": [],
   "source": [
    "# SELECT columns\n",
    "# select employee_id, name, age, salary from emp\n",
    "\n",
    "emp_filtered = emp.select(col(\"employee_id\"), expr(\"name\"), emp.age, emp.salary)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "585c81ff-0ffd-4ef8-9a4f-24fcafab93fd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------------+---+------+\n",
      "|employee_id|         name|age|salary|\n",
      "+-----------+-------------+---+------+\n",
      "|        001|     John Doe| 30| 50000|\n",
      "|        002|   Jane Smith| 25| 45000|\n",
      "|        003|    Bob Brown| 35| 55000|\n",
      "|        004|    Alice Lee| 28| 48000|\n",
      "|        005|    Jack Chan| 40| 60000|\n",
      "|        006|    Jill Wong| 32| 52000|\n",
      "|        007|James Johnson| 42| 70000|\n",
      "|        008|     Kate Kim| 29| 51000|\n",
      "|        009|      Tom Tan| 33| 58000|\n",
      "|        010|     Lisa Lee| 27| 47000|\n",
      "|        011|   David Park| 38| 65000|\n",
      "|        012|   Susan Chen| 31| 54000|\n",
      "|        013|    Brian Kim| 45| 75000|\n",
      "|        014|    Emily Lee| 26| 46000|\n",
      "|        015|  Michael Lee| 37| 63000|\n",
      "|        016|  Kelly Zhang| 30| 49000|\n",
      "|        017|  George Wang| 34| 57000|\n",
      "|        018|    Nancy Liu| 29| 50000|\n",
      "|        019|  Steven Chen| 36| 62000|\n",
      "|        020|    Grace Kim| 32| 53000|\n",
      "+-----------+-------------+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SHOW Dataframe (ACTION)\n",
    "\n",
    "emp_filtered.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "13a83875-2022-4c1b-8845-dce7121767d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Using expr for select\n",
    "# select employee_id as emp_id, name, cast(age as int) as age, salary from emp_filtered\n",
    "\n",
    "emp_casted = emp_filtered.select(expr(\"employee_id as emp_id\"), emp.name, expr(\"cast(age as int) as age\"), emp.salary)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "40a102d5-88bb-4183-ab16-a04a01f727d1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------------+---+------+\n",
      "|emp_id|         name|age|salary|\n",
      "+------+-------------+---+------+\n",
      "|   001|     John Doe| 30| 50000|\n",
      "|   002|   Jane Smith| 25| 45000|\n",
      "|   003|    Bob Brown| 35| 55000|\n",
      "|   004|    Alice Lee| 28| 48000|\n",
      "|   005|    Jack Chan| 40| 60000|\n",
      "|   006|    Jill Wong| 32| 52000|\n",
      "|   007|James Johnson| 42| 70000|\n",
      "|   008|     Kate Kim| 29| 51000|\n",
      "|   009|      Tom Tan| 33| 58000|\n",
      "|   010|     Lisa Lee| 27| 47000|\n",
      "|   011|   David Park| 38| 65000|\n",
      "|   012|   Susan Chen| 31| 54000|\n",
      "|   013|    Brian Kim| 45| 75000|\n",
      "|   014|    Emily Lee| 26| 46000|\n",
      "|   015|  Michael Lee| 37| 63000|\n",
      "|   016|  Kelly Zhang| 30| 49000|\n",
      "|   017|  George Wang| 34| 57000|\n",
      "|   018|    Nancy Liu| 29| 50000|\n",
      "|   019|  Steven Chen| 36| 62000|\n",
      "|   020|    Grace Kim| 32| 53000|\n",
      "+------+-------------+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SHOW Dataframe (ACTION)\n",
    "\n",
    "emp_casted.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "4ca5ae3a-4298-4ef2-b599-02207633ccf8",
   "metadata": {},
   "outputs": [],
   "source": [
    "emp_casted_1 = emp_filtered.selectExpr(\"employee_id as emp_id\", \"name\", \"cast(age as int) as age\", \"salary\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "738c6b74-2307-472b-8ef3-9c05427d49b3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------------+---+------+\n",
      "|emp_id|         name|age|salary|\n",
      "+------+-------------+---+------+\n",
      "|   001|     John Doe| 30| 50000|\n",
      "|   002|   Jane Smith| 25| 45000|\n",
      "|   003|    Bob Brown| 35| 55000|\n",
      "|   004|    Alice Lee| 28| 48000|\n",
      "|   005|    Jack Chan| 40| 60000|\n",
      "|   006|    Jill Wong| 32| 52000|\n",
      "|   007|James Johnson| 42| 70000|\n",
      "|   008|     Kate Kim| 29| 51000|\n",
      "|   009|      Tom Tan| 33| 58000|\n",
      "|   010|     Lisa Lee| 27| 47000|\n",
      "|   011|   David Park| 38| 65000|\n",
      "|   012|   Susan Chen| 31| 54000|\n",
      "|   013|    Brian Kim| 45| 75000|\n",
      "|   014|    Emily Lee| 26| 46000|\n",
      "|   015|  Michael Lee| 37| 63000|\n",
      "|   016|  Kelly Zhang| 30| 49000|\n",
      "|   017|  George Wang| 34| 57000|\n",
      "|   018|    Nancy Liu| 29| 50000|\n",
      "|   019|  Steven Chen| 36| 62000|\n",
      "|   020|    Grace Kim| 32| 53000|\n",
      "+------+-------------+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "emp_casted_1.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "b947ee1b-900d-40ad-b3ac-4fb18cac31d8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- emp_id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      " |-- salary: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "emp_casted.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "46f92cc3-258c-4d31-9403-546902cfacf4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Filter emp based on Age > 30\n",
    "# select emp_id, name, age, salary from emp_casted where age > 30\n",
    "\n",
    "emp_final = emp_casted.select(\"emp_id\", \"name\", \"age\", \"salary\").where(\"age > 30\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "5e26d7ad-a380-4290-b8eb-31e3c84a425d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------------+---+------+\n",
      "|emp_id|         name|age|salary|\n",
      "+------+-------------+---+------+\n",
      "|   003|    Bob Brown| 35| 55000|\n",
      "|   005|    Jack Chan| 40| 60000|\n",
      "|   006|    Jill Wong| 32| 52000|\n",
      "|   007|James Johnson| 42| 70000|\n",
      "|   009|      Tom Tan| 33| 58000|\n",
      "|   011|   David Park| 38| 65000|\n",
      "|   012|   Susan Chen| 31| 54000|\n",
      "|   013|    Brian Kim| 45| 75000|\n",
      "|   015|  Michael Lee| 37| 63000|\n",
      "|   017|  George Wang| 34| 57000|\n",
      "|   019|  Steven Chen| 36| 62000|\n",
      "|   020|    Grace Kim| 32| 53000|\n",
      "+------+-------------+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SHOW Dataframe (ACTION)\n",
    "\n",
    "emp_final.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "a38b9d29-c225-44c4-9cf5-29fb1e35fe83",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write the data back as CSV (ACTION)\n",
    "\n",
    "emp_final.write.format(\"csv\").save(\"data/output/2/emp.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "c6c5dee2-e504-4f31-bf38-482bd0a7f7fa",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType([StructField('name', StringType(), True), StructField('age', IntegerType(), True)])"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Bonus TIP\n",
    "\n",
    "schema_str = \"name string, age int\"\n",
    "\n",
    "from pyspark.sql.types import _parse_datatype_string\n",
    "\n",
    "schema_spark = _parse_datatype_string(schema_str)\n",
    "\n",
    "schema_spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dababad5-cb20-4387-9ac2-0682f036871d",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}