Uber1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cc21166d",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.storagelevel import StorageLevel\n",
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "635e94b3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# configure Spark\n",
    "conf = SparkConf().setAppName(\"MyApp\") \\\n",
    "                  .setMaster(\"local[*]\") \\\n",
    "                  .set(\"spark.executor.memory\", \"4g\") \\\n",
    "                  .set(\"spark.driver.memory\", \"4g\") \\\n",
    "                  .set(\"spark.executor.cores\", \"2\")\n",
    "# sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "da1a6f6d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create Spark session\n",
    "spark = SparkSession.builder \\\n",
    "    .config(conf=conf) \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "73ab0434",
   "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://DESKTOP-LDP3CPA.mshome.net:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.3.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>MyApp</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x2751593cb20>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "7b1362f1",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set(\"spark.sql.legacy.timeParserPolicy\", \"LEGACY\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "5d68dbec",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read the uber file into a pyspark dataframe\n",
    "uber1 = spark.read.csv(\"C:\\\\Users\\\\USER\\\\TransTech Project\\\\uber-raw-data-apr14.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "ce1db11e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read the uber file into a pyspark dataframe\n",
    "base = spark.read.csv(\"C:\\\\Users\\\\USER\\\\TransTech Project\\\\TLC Affiliated Base.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "d8e32267",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Change the names of the columns to align with the ones in other datasets\n",
    "columns_mapping = {\"Date/Time\": \"Date_time\", \"Lat\": \"Latitude\", \"Lon\": \"Longitude\"}\n",
    "\n",
    "for old_name, new_name in columns_mapping.items():\n",
    "    uber1 = uber1.withColumnRenamed(old_name, new_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "b17bfea8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# split the date_time columns\n",
    "uber1 = uber1.withColumn(\"DATE_TIME\", split(uber1[\"DATE_TIME\"], \" \"))\n",
    "uber1 = uber1.withColumn(\"Date\", uber1[\"DATE_TIME\"].getItem(0))\n",
    "uber1 = uber1.withColumn(\"Time\", uber1[\"DATE_TIME\"].getItem(1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "4063029f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# drop irrelevant columns\n",
    "uber1 = uber1.drop(\"DATE_TIME\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "95967a06",
   "metadata": {},
   "outputs": [],
   "source": [
    "# convert data type of DATE column to Date datatype\n",
    "uber1 = uber1.withColumn(\"Date\", to_date(uber1[\"Date\"], \"MM/dd/yyyy\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "b0261b45",
   "metadata": {},
   "outputs": [],
   "source": [
    "# convert the \"time_column\" from string to 24 hour time format\n",
    "uber1 = uber1.withColumn(\"Time\", col(\"Time\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "1293ad48",
   "metadata": {},
   "outputs": [],
   "source": [
    "# cast longitude and latitude columns to double_precision\n",
    "uber1 = uber1.withColumn(\"Latitude\", col(\"Latitude\").cast(\"double\")) \\\n",
    "                           .withColumn(\"Longitude\", col(\"Longitude\").cast(\"double\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "52095ff9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add a new column to the dataset\n",
    "uber1 = uber1.withColumn(\"Company\", lit(\"Uber\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "354d0f46",
   "metadata": {},
   "outputs": [],
   "source": [
    "# join the uber dataframe to the base dataframe\n",
    "uber1 = uber1.join(base, uber1.Base== base.Base_Code)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "878586fa",
   "metadata": {},
   "outputs": [],
   "source": [
    "# rearrange the columns\n",
    "uber1 = uber1.selectExpr(\"Date\", \"Time\", \"Company\", \"Base_Code\", \"Base_Name\", \"Base_Region\", \"Latitude\", \"Longitude\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "220f2ff1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# remove blank cells\n",
    "uber1 = uber1.dropna()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "1a5fd258",
   "metadata": {},
   "outputs": [],
   "source": [
    "# drop duplicates\n",
    "uber1 = uber1.dropDuplicates()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "7995c21b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Date: date, Time: string, Company: string, Base_Code: string, Base_Name: string, Base_Region: string, Latitude: double, Longitude: double]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Cache the table\n",
    "uber1.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "5536aba2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# write to postgresql server database\n",
    "uber1.write \\\n",
    "  .format(\"jdbc\") \\\n",
    "  .option(\"url\", \"jdbc:postgresql://localhost:5432/postgis_33_sample\") \\\n",
    "  .option(\"dbtable\", \"uber2014\") \\\n",
    "  .option(\"user\", \"postgres\") \\\n",
    "  .option(\"password\", \"*********\") \\\n",
    "  .option(\"batchsize\", 1000) \\\n",
    "  .mode(\"append\") \\\n",
    "  .save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cde2fb2a",
   "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.9.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}