deliverable2_database.py

# -*- coding: utf-8 -*-
"""deliverable2_database.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1vMllL4GElvMxmiNt_ct0UrE0GpvU9j-1
"""

import os
# Find the latest version of spark 3.0  from http://www-us.apache.org/dist/spark/ and enter as the spark version
# For example:
# spark_version = 'spark-3.0.1'
spark_version = 'spark-3.0.1'
os.environ['SPARK_VERSION']=spark_version

# Install Spark and Java
!apt-get update
!apt-get install openjdk-11-jdk-headless -qq > /dev/null
!wget -q http://www-us.apache.org/dist/spark/$SPARK_VERSION/$SPARK_VERSION-bin-hadoop2.7.tgz
!tar xf $SPARK_VERSION-bin-hadoop2.7.tgz
!pip install -q findspark

# Set Environment Variables
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-11-openjdk-amd64"
os.environ["SPARK_HOME"] = f"/content/{spark_version}-bin-hadoop2.7"

# Start a SparkSession
import findspark
findspark.init()

!wget https://jdbc.postgresql.org/download/postgresql-42.2.9.jar

# Start Spark session
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ETL").config("spark.driver.extraClassPath","/content/postgresql-42.2.9.jar").getOrCreate()

# Read in raw crash CSV

df_austin_crash_data = spark.read.option('header', 'true').csv("2018-2020_Austin_crash_data.csv", inferSchema=True, sep=',', timestampFormat="mm/dd/yy")
df_austin_crash_data.show(10)

# Read in car ratings CSV

df_car_data = spark.read.option('header', 'true').csv("car_data.csv", inferSchema=True, sep=',', timestampFormat="mm/dd/yy").drop("_c0")
df_car_data.show(10)

from rds import db_password

# Write DataFrame to RDS
mode="overwrite"
jdbc_url = "jdbc:postgresql://final-project.cn1djdbx7lfi.us-east-1.rds.amazonaws.com:5432/car_db"
config = {"user":"postgres",
          "password": {db_password},
          "driver":"org.postgresql.Driver"}

# Write raw_data table to RDBS
df_austin_crash_data.write.jdbc(url=jdbc_url, table='raw_data', mode=mode, properties=config)

df_car_data.count()

df_austin_crash_data.count()

# Remove duplicates from car ratings
df_car_data = df_car_data.drop_duplicates()

df_car_data.count()

# Merge raw data and car ratings - left join 
df = df_austin_crash_data.join(df_car_data, on=["Vehicle Make", "Vehicle Model Name", "Vehicle Model Year"], how="left")
df.show()

df.count()

# Write merged table to RDBS
df.write.jdbc(url=jdbc_url, table='total_crash_data', mode=mode, properties=config)

# Write car ratings table to RDBS
df_car_data.write.jdbc(url=jdbc_url, table='car_safety_ratings', mode=mode, properties=config)