# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# + [markdown] colab_type="text" id="view-in-github"
# <a href="https://colab.research.google.com/github/weihaolun/Twitter-Sentiment-Analysis/blob/main/Database/ETL.ipynb" target="_parent"><img src="https://colab.research.google.com/assets/colab-badge.svg" alt="Open In Colab"/></a>

# + colab={"base_uri": "https://localhost:8080/"} id="V58rxea0HqSa" outputId="3ef04a64-0697-46d3-b601-b009384ee93f"
import os
# Find the latest version of spark 3.0 from http://www.apache.org/dist/spark/ and enter as the spark version
spark_version = 'spark-3.0.3'
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.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
import os
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()

# + colab={"base_uri": "https://localhost:8080/"} id="_xKwTpATHqSe" outputId="153c5bdb-161a-4714-cd00-98cdf0b967a9"
# Download the Postgres driver that will allow Spark to interact with Postgres.
# !wget https://jdbc.postgresql.org/download/postgresql-42.2.16.jar

# + id="MMqDAjVS0KN9"
# Create spark session
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("final-project").config("spark.driver.extraClassPath","/content/postgresql-42.2.16.jar").getOrCreate()

# + [markdown] id="cyBsySGuY-9V"
# ### Load Amazon Data into Spark DataFrames

# + colab={"base_uri": "https://localhost:8080/"} id="CtCmBhQJY-9Z" outputId="bc4cbb5c-396d-4881-cfdf-df0e1d45cf87"
from pyspark import SparkFiles
# positive tweets 
url = "https://sjpv94twitter.s3.us-west-1.amazonaws.com/positive.csv"
spark.sparkContext.addFile(url)
positive_df = spark.read.option("encoding", "UTF-8").csv(SparkFiles.get("positive.csv"), sep=",", header=True, inferSchema=True)
positive_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="ECM5F_qHKskH" outputId="97cd408c-af02-44b1-f3dc-8fd46f87d4e1"
# negative tweets 
url = "https://sjpv94twitter.s3.us-west-1.amazonaws.com/negative.csv"
spark.sparkContext.addFile(url)
negative_df = spark.read.option("encoding", "UTF-8").csv(SparkFiles.get("negative.csv"), sep=",", header=True, inferSchema=True)
negative_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="iNyq8mX_KtpC" outputId="e150a99e-0b11-4e3f-80a6-5a9e75609994"
# neutral tweets 
url = "https://sjpv94twitter.s3.us-west-1.amazonaws.com/neutral.csv"
spark.sparkContext.addFile(url)
neutral_df = spark.read.option("encoding", "UTF-8").csv(SparkFiles.get("neutral.csv"), sep=",", header=True, inferSchema=True)
neutral_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="irYD-HhqPH5U" outputId="9fec7845-265c-4759-a703-4db63c133c5d"
# api tweets 
url = "https://sjpv94twitter.s3.us-west-1.amazonaws.com/cleaned_scored_tweets.json"
spark.sparkContext.addFile(url)
api_tweets_df = spark.read.option("encoding", "UTF-8").json(SparkFiles.get("cleaned_scored_tweets.json"))
api_tweets_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="TEXIgPmmPIIA" outputId="24ae092d-387b-43fd-99e9-3baef901ac40"
# api count
url = "https://sjpv94twitter.s3.us-west-1.amazonaws.com/tweets_count.json"
spark.sparkContext.addFile(url)
api_count_df = spark.read.option("encoding", "UTF-8").json(SparkFiles.get("tweets_count.json"))
api_count_df.show()

# + [markdown] id="2yUSe55VY-9t"
# ### Create DataFrames to match tables in pgAdmin

# + colab={"base_uri": "https://localhost:8080/"} id="fhwiBX-KMNH0" outputId="8043445c-9569-4392-9c30-8d971f8481eb"
# combine positive, negative, neutral tweets to match tables in Postgres
initial_tweets = positive_df.unionByName(negative_df)
initial_tweets = initial_tweets.unionByName(neutral_df)
initial_tweets = initial_tweets.drop(initial_tweets._c0)
initial_tweets.show()

# + colab={"base_uri": "https://localhost:8080/"} id="J0fvnVRdd77c" outputId="a3836dc5-340e-49c0-8f10-47488f28bab7"
# rearrange columns
from pyspark.sql.functions import col
from pyspark.sql.functions import to_date
api_tweets_df = api_tweets_df.select(["created_at", to_date("created_date", 'yyyy-MM-dd').alias("created_date"), "weekday",\
                                      col("hours").cast("int"), "tweet", "text", "score"])
api_tweets_df.show()

# + colab={"base_uri": "https://localhost:8080/"} id="9iMpot1Tgi5b" outputId="6ec3cd65-64bc-4ca1-beb6-f3ea161de807"
# rearrange columns
api_count_df = api_count_df.select(["end_time","start_time",to_date("created_date", 'yyyy-MM-dd').alias("created_date"),"weekday","tweet_count"])
api_count_df.show()

# + [markdown] id="jITZhLkmY--J"
# ### Connect to the AWS RDS instance and write each DataFrame to its table. 

# + colab={"base_uri": "https://localhost:8080/"} id="7jiUvs1aY--L" outputId="bfcebb78-2588-4503-b058-f1b1bd524c90"
# Configure settings for RDS
from getpass import getpass
password = getpass('<PASSWORD>')

mode = "append"
jdbc_url="jdbc:postgresql://<connection string>/<dbname>"
config = {"user":"postgres", 
          "password": password, 
          "driver":"org.postgresql.Driver"}

# Deleted connection string, and db name
# For the purpose of this project,
# connection string used was 'twitterproject.cgrabvwy3u7o.us-west-1.rds.amazonaws.com:5432'
# db name used was 'twitterproject'
# Database will be deleted after project submission

# + id="T2zgZ-aKY--Q"
# Write initial_tweets to table in RDS
initial_tweets.write.jdbc(url=jdbc_url, table='initial_tweets_table', mode=mode, properties=config)

# + id="KbXri15fY--Z"
# Write api_tweets_df to table in RDS
api_tweets_df.write.jdbc(url=jdbc_url, table='api_tweets_table', mode=mode, properties=config)

# + id="XdQknSHLY--e"
# Write api_count_df to table in RDS
api_count_df.write.jdbc(url=jdbc_url, table='api_count_table', mode=mode, properties=config)

# + id="Exuo6ebUsCqW"


StarcoderdataJupyterScriptsDedupFiltered