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

Automatically generated by Colaboratory.

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

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

# Set Environment Variables
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.0.1-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("CloudETL").config("spark.driver.extraClassPath","/content/postgresql-42.2.9.jar").getOrCreate()

# Import struct fields that we can use
from pyspark.sql.types import StructField, StringType, IntegerType, StructType, LongType, FloatType

# Creating the list of struct fields
schema = [StructField("county_name", StringType(), False), StructField("county_code", IntegerType(), True), StructField("date", StringType(), True), StructField("aqi", IntegerType(), True), StructField("category", StringType(), True), StructField("defining_parameter", StringType(), True)]
schema

# Pass in our AQ fields
aq_final = StructType(fields=schema)
aq_final

# Read in air quality data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/aq_dataframe.csv"
spark.sparkContext.addFile(url)
aq_df = spark.read.csv(SparkFiles.get("aq_dataframe.csv"), schema=aq_final, sep=",", header=True)

# Show DataFrame
aq_df.printSchema()

# Create a list of struct fields for Hummingbirds
hb_schema = [StructField("genus", StringType(), True), 
             StructField("species", StringType(), True), 
             StructField("country_code", StringType(), True), 
             StructField("locality", StringType(), True), 
             StructField("state", StringType(), True), 
             StructField("occurence_status", StringType(), True), 
             StructField("individual_count", LongType(), True), 
             StructField("latitude", FloatType(), True),
             StructField("longitude", FloatType(), True),
             StructField("basis_of_record", StringType(), True),
             StructField("catalog_number", StringType(), True),
             StructField("date", StringType(), True)]

# Pass in our HB fields
hb_final = StructType(fields=hb_schema)
hb_final

# Read in ebirds data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/hb_dataframe.csv"
spark.sparkContext.addFile(url)
hb_df = spark.read.csv(SparkFiles.get("hb_dataframe.csv"), schema=hb_final, sep=",", header=True)

# Show DataFrame
hb_df.printSchema()

# Creating the list of struct fields
c_schema = [StructField("latitude", FloatType(), True), StructField("longitude", FloatType(), True), StructField("county_name", StringType(), True)]
c_schema

# Pass in our County fields
c_final = StructType(fields=c_schema)
c_final

# Read in counties data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/counties_final.csv"
spark.sparkContext.addFile(url)
counties_df = spark.read.csv(SparkFiles.get("counties_final.csv"), schema=c_final, sep=",", header=True)

# Show DataFrame
counties_df.printSchema()

# Creating the list of struct fields
w_schema = [StructField("county", StringType(), True), StructField("date", StringType(), True), StructField("high_wind", StringType(), True), StructField("heavy_rain", StringType(), True)]
w_schema

# Pass in our W fields
w_final = StructType(fields=w_schema)
w_final

# Creating the list of struct fields
cp_schema = [StructField("county", StringType(), True), StructField("year", StringType(), True), StructField("population", StringType(), True)]
cp_schema

# Pass in our CP fields
cp_final = StructType(fields=cp_schema)
cp_final

# Read in weather data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/wind_rain.csv"
spark.sparkContext.addFile(url)
weather_df = spark.read.csv(SparkFiles.get("wind_rain.csv"), schema=w_final, sep=",", header=True)

# Show DataFrame
weather_df.printSchema()

# Read in county pop data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/population_county.csv"
spark.sparkContext.addFile(url)
county_pop_df = spark.read.csv(SparkFiles.get("population_county.csv"), schema=cp_final, sep=",", header=True)

# Show DataFrame
county_pop_df.printSchema()

county_pop_df.head()

# Configure settings for RDS
mode = "append"
jdbc_url="jdbc:postgresql://aves.cos0wnwxlodh.us-east-2.rds.amazonaws.com:5432/aves"
config = {"user":"caznoe", 
          "password": "<PASSWORD>", 
          "driver":"org.postgresql.Driver"}

# Write DataFrame to weather table in RDS
weather_df.write.jdbc(url=jdbc_url, table='weather', mode=mode, properties=config)

# Write DataFrame to air quality table in RDS
aq_df.write.jdbc(url=jdbc_url, table='air_quality', mode=mode, properties=config)

# Write DataFrame to ebirds table in RDS
hb_df.write.jdbc(url=jdbc_url, table='ebirds', mode=mode, properties=config)

# Write DataFrame to counties table in RDS
counties_df.write.jdbc(url=jdbc_url, table='counties', mode=mode, properties=config)

# Write DataFrame to county population table in RDS
county_pop_df.write.jdbc(url=jdbc_url, table='county_pop', mode=mode, properties=config)
# -*- coding: utf-8 -*-
"""aves.ipynb

Automatically generated by Colaboratory.

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

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

# Set Environment Variables
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.0.1-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("CloudETL").config("spark.driver.extraClassPath","/content/postgresql-42.2.9.jar").getOrCreate()

# Import struct fields that we can use
from pyspark.sql.types import StructField, StringType, IntegerType, StructType, LongType, FloatType

# Creating the list of struct fields
schema = [StructField("county_name", StringType(), False), StructField("county_code", IntegerType(), True), StructField("date", StringType(), True), StructField("aqi", IntegerType(), True), StructField("category", StringType(), True), StructField("defining_parameter", StringType(), True)]
schema

# Pass in our AQ fields
aq_final = StructType(fields=schema)
aq_final

# Read in air quality data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/aq_dataframe.csv"
spark.sparkContext.addFile(url)
aq_df = spark.read.csv(SparkFiles.get("aq_dataframe.csv"), schema=aq_final, sep=",", header=True)

# Show DataFrame
aq_df.printSchema()

# Create a list of struct fields for Hummingbirds
hb_schema = [StructField("genus", StringType(), True), 
             StructField("species", StringType(), True), 
             StructField("country_code", StringType(), True), 
             StructField("locality", StringType(), True), 
             StructField("state", StringType(), True), 
             StructField("occurence_status", StringType(), True), 
             StructField("individual_count", LongType(), True), 
             StructField("latitude", FloatType(), True),
             StructField("longitude", FloatType(), True),
             StructField("basis_of_record", StringType(), True),
             StructField("catalog_number", StringType(), True),
             StructField("date", StringType(), True)]

# Pass in our HB fields
hb_final = StructType(fields=hb_schema)
hb_final

# Read in ebirds data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/hb_dataframe.csv"
spark.sparkContext.addFile(url)
hb_df = spark.read.csv(SparkFiles.get("hb_dataframe.csv"), schema=hb_final, sep=",", header=True)

# Show DataFrame
hb_df.printSchema()

# Creating the list of struct fields
c_schema = [StructField("latitude", FloatType(), True), StructField("longitude", FloatType(), True), StructField("county_name", StringType(), True)]
c_schema

# Pass in our County fields
c_final = StructType(fields=c_schema)
c_final

# Read in counties data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/counties_final.csv"
spark.sparkContext.addFile(url)
counties_df = spark.read.csv(SparkFiles.get("counties_final.csv"), schema=c_final, sep=",", header=True)

# Show DataFrame
counties_df.printSchema()

# Creating the list of struct fields
w_schema = [StructField("county", StringType(), True), StructField("date", StringType(), True), StructField("high_wind", StringType(), True), StructField("heavy_rain", StringType(), True)]
w_schema

# Pass in our W fields
w_final = StructType(fields=w_schema)
w_final

# Creating the list of struct fields
cp_schema = [StructField("county", StringType(), True), StructField("year", StringType(), True), StructField("population", StringType(), True)]
cp_schema

# Pass in our CP fields
cp_final = StructType(fields=cp_schema)
cp_final

# Read in weather data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/wind_rain.csv"
spark.sparkContext.addFile(url)
weather_df = spark.read.csv(SparkFiles.get("wind_rain.csv"), schema=w_final, sep=",", header=True)

# Show DataFrame
weather_df.printSchema()

# Read in county pop data from S3 Buckets
from pyspark import SparkFiles
url ="https://caznoe-aves.s3.amazonaws.com/population_county.csv"
spark.sparkContext.addFile(url)
county_pop_df = spark.read.csv(SparkFiles.get("population_county.csv"), schema=cp_final, sep=",", header=True)

# Show DataFrame
county_pop_df.printSchema()

county_pop_df.head()

# Configure settings for RDS
mode = "append"
jdbc_url="jdbc:postgresql://aves.cos0wnwxlodh.us-east-2.rds.amazonaws.com:5432/aves"
config = {"user":"caznoe", 
          "password": "<PASSWORD>", 
          "driver":"org.postgresql.Driver"}

# Write DataFrame to weather table in RDS
weather_df.write.jdbc(url=jdbc_url, table='weather', mode=mode, properties=config)

# Write DataFrame to air quality table in RDS
aq_df.write.jdbc(url=jdbc_url, table='air_quality', mode=mode, properties=config)

# Write DataFrame to ebirds table in RDS
hb_df.write.jdbc(url=jdbc_url, table='ebirds', mode=mode, properties=config)

# Write DataFrame to counties table in RDS
counties_df.write.jdbc(url=jdbc_url, table='counties', mode=mode, properties=config)

# Write DataFrame to county population table in RDS
county_pop_df.write.jdbc(url=jdbc_url, table='county_pop', mode=mode, properties=config)
# -*- coding: utf-8 -*- aves.ipynb  Automatically generated by Colaboratory.  Original file is located at     https://colab.research.google.com/drive/1B0dqujGig62wioC2lIlRG8tiUCoWRnCy # Install Java, Spark, and Findspark # Set Environment Variables # Start a SparkSession # Start Spark Session # Import struct fields that we can use # Creating the list of struct fields # Pass in our AQ fields # Read in air quality data from S3 Buckets # Show DataFrame # Create a list of struct fields for Hummingbirds # Pass in our HB fields # Read in ebirds data from S3 Buckets # Show DataFrame # Creating the list of struct fields # Pass in our County fields # Read in counties data from S3 Buckets # Show DataFrame # Creating the list of struct fields # Pass in our W fields # Creating the list of struct fields # Pass in our CP fields # Read in weather data from S3 Buckets # Show DataFrame # Read in county pop data from S3 Buckets # Show DataFrame # Configure settings for RDS # Write DataFrame to weather table in RDS # Write DataFrame to air quality table in RDS # Write DataFrame to ebirds table in RDS # Write DataFrame to counties table in RDS # Write DataFrame to county population table in RDS