<jupyter_start><jupyter_text><jupyter_code>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()
# Download the Postgres driver that will allow Spark to interact with Postgres.
!wget https://jdbc.postgresql.org/download/postgresql-42.2.16.jar
# 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()<jupyter_output><empty_output><jupyter_text>Load Amazon Data into Spark DataFrames<jupyter_code>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()
# 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()
# 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()
# 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()
# 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()<jupyter_output>+------------+--------------------+--------------------+-----------+---------+
|created_date|            end_time|          start_time|tweet_count|  weekday|
+------------+--------------------+--------------------+-----------+---------+
|  2021-11-01|2021-11-02T00:00:...|2021-11-01T00:00:...|     138498|   Monday|
|  2021-11-02|2021-11-03T00:00:...|2021-11-02T00:00:...|      88621|  Tuesday|
|  2021-11-03|2021-11-04T00:00:...|2021-11-03T00:00:...|      53148|Wednesday|
|  2021-11-04|2021-11-05T00:00:...|2021-11-04T00:00:...|      54606| Thursday|
|  2021-11-05|2021-11-06T00:00:...|2021-11-05T00:00:...|      47726|   Friday|
|  2021-11-06|2021-11-07T00:00:...|2021-11-06T00:00:...|      64812| Saturday|
|  2021-11-07|2021-11-08T00:00:...|2021-11-07T00:00:...|      78984|   Sunday|
|  2021-11-08|2021-11-09T00:00:...|2021-11-08T00:00:...|      74188|   Monday|
|  2021-11-09|2021-11-10T00:00:...|2021-11-09T00:00:...|      51366|  Tuesday|
|  2021-11-10|2021-11-11T00:00:...|2021-11-10T00:00:[...]<jupyter_text>Create DataFrames to match tables in pgAdmin<jupyter_code># 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()
# 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()
# 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()<jupyter_output>+--------------------+--------------------+------------+---------+-----------+
|            end_time|          start_time|created_date|  weekday|tweet_count|
+--------------------+--------------------+------------+---------+-----------+
|2021-11-02T00:00:...|2021-11-01T00:00:...|  2021-11-01|   Monday|     138498|
|2021-11-03T00:00:...|2021-11-02T00:00:...|  2021-11-02|  Tuesday|      88621|
|2021-11-04T00:00:...|2021-11-03T00:00:...|  2021-11-03|Wednesday|      53148|
|2021-11-05T00:00:...|2021-11-04T00:00:...|  2021-11-04| Thursday|      54606|
|2021-11-06T00:00:...|2021-11-05T00:00:...|  2021-11-05|   Friday|      47726|
|2021-11-07T00:00:...|2021-11-06T00:00:...|  2021-11-06| Saturday|      64812|
|2021-11-08T00:00:...|2021-11-07T00:00:...|  2021-11-07|   Sunday|      78984|
|2021-11-09T00:00:...|2021-11-08T00:00:...|  2021-11-08|   Monday|      74188|
|2021-11-10T00:00:...|2021-11-09T00:00:...|  2021-11-09|  Tuesday|      51366|
|2021-11-11T00:00:...|2021-11-10T00:00:...|  2021-11[...]<jupyter_text>Connect to the AWS RDS instance and write each DataFrame to its table.<jupyter_code># Configure settings for RDS
from getpass import getpass
password = getpass('Enter database 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
# Write initial_tweets to table in RDS
initial_tweets.write.jdbc(url=jdbc_url, table='initial_tweets_table', mode=mode, properties=config)
# Write api_tweets_df to table in RDS
api_tweets_df.write.jdbc(url=jdbc_url, table='api_tweets_table', mode=mode, properties=config)
# Write api_count_df to table in RDS
api_count_df.write.jdbc(url=jdbc_url, table='api_count_table', mode=mode, properties=config)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.21385402138540213, 'fraction_numerical': 0.12242367890903456, 'mean_word_length': 3.6364942528735633, 'pattern_counts': {'":': 3, '<': 17, '<?xml version=': 0, '>': 18, 'https://': 6, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '453749', 'license': 'MIT', 'n_tokens_mistral': 2921, 'n_tokens_neox': 2464, 'n_words': 353, 'path': 'Database/ETL.ipynb', 'repo_name': 'weihaolun/Twitter-Sentiment-Analysis', 'size': '6453'}