import os
import sys

os.environ["SPARK_HOME"] = "/Users/projects/.pyenv/versions/3.7.10/envs/tatapower/lib/python3.7/site-packages/pyspark"
# os.environ["HADOOP_HOME"] = ""
# os.environ["PYSPARK_PYTHON"] = "/opt/cloudera/parcels/Anaconda/bin/python"
# os.environ["JAVA_HOME"] = "/usr/java/jdk1.8.0_161/jre"
# os.environ["PYLIB"] = os.environ["SPARK_HOME"] + "/python/lib"
# sys.path.insert(0, os.environ["PYLIB"] +"/py4j-0.10.6-src.zip")
# sys.path.insert(0, os.environ["PYLIB"] +"/pyspark.zip")

"""
Explanation: ETL with PySpark SQL
End of explanation
"""


from pyspark import SparkFiles
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.types import *
from pyspark.sql.functions import *

# Create a SparkSession.
spark = SparkSession.builder\
                .master("local[*]")\
                .appName("ETL")\
                .config("spark.executor.logs.rolling.time.interval", "daily")\
                .getOrCreate()


"""
Explanation: Importing and creating SparkSession
End of explanation
"""


# Social Network edgelist

# First, we set the filename
file = "HiggsTwitter/higgs-social_network.edgelist.gz"

# Second, Set the Schema where first column is follower and second is followed, both of types integer.
schema = StructType([StructField("follower", IntegerType()), StructField("followed", IntegerType())])

# Create the DataFrame
socialDF = spark.read.csv(path=file, sep=" ", schema=schema)

#Retweet Network

# First, we set the filename
file = "HiggsTwitter/higgs-retweet_network.edgelist.gz"

# Second, Set the Schema where first column is tweeter, second is tweeted, third is occur and all are of type integer.
schema = 

# Create the DataFrame
retweetDF =

# Reply Network

# First, we set the filename
file = "HiggsTwitter/higgs-reply_network.edgelist.gz"

# Second, Set the Schema where first column is replier, second is replied, third is occur and all are of type integer.
schema = 

# Create the DataFrame
replyDF = 


# Mention Network

# First, we set the filename
file = "HiggsTwitter/higgs-mention_network.edgelist.gz"

# Second, Set the Schema where first column is mentioner, second is mentioned, third is occur and all are of type integer.
schema = 

# Create the DataFrame
mentionDF = 

# Activity Time

# First, we set the filename
file = "HiggsTwitter/higgs-activity_time.txt.gz"

# Second, Set the Schema where 
#    * first column is userA (integer)
#    * second is userB (integer)
#    * third is timestamp (integer)
#    * fourth is interaction (string): Interaction can be: RT (retweet), MT (mention) or RE (reply)
schema = 
activityDF = 

"""
Explanation: Setting filesystem and files
Load all CSV's files from HiggsTwitter dataset (http://snap.stanford.edu/data/higgs-twitter.html)
Read all the 5 different zip files into Spark Dataframe.
End of explanation
"""


# Save all the five files to parquet format

socialDF
retweetDF
replyDF
mentionDF
activityDF

"""
Explanation: Convert CSV's dataframes to Apache Parquet files
End of explanation
"""


# Read all the five files from parquet format

socialDFpq = spark.read
retweetDFpq = spark.read
replyDFpq = spark.read
mentionDFpq = spark.read
activityDFpq = spark.read

"""
Explanation: Load the parquet files into new dataframes
End of explanation
"""


# Display the schema of the dataframes

socialDFpq
socialDFpq

# Show the top 5 rows of each dataframe

socialDFpq
retweetDFpq
replyDFpq
mentionDFpq
activityDFpq

"""
Explanation: Working with dataframes
End of explanation
"""


# Users who have most followers

socialDFpq

# Users who have most mentions
mentionDFpq

# Of the top 5 followed users, how many mentions has each one?

# top_f contains "top 5 users who have most followers"
top_f = 

top_f.

"""
Explanation: Spark SQL using DataFrames API
End of explanation
"""


# Create temporary views so we can use SQL statements
socialDFpq.
retweetDFpq.
replyDFpq.
mentionDFpq.
activityDFpq.

# List all the tables in spark memory

spark.

# Users who have most followers using SQL
spark.


# Users who have most mentions using SQL
spark.

# Of the top 5 followed users, how many mentions has each one? (using SQL)


"""
Explanation: Spark SQL using SQL language
Create temporary views of all tables so we can use SQL statements
End of explanation
"""


%%time
# GZIP Compressed CSV
socialDF.groupBy("followed").agg(count("follower").alias("followers")).orderBy(desc("followers")).show(5)

%%time
# Parquet file
socialDFpq.groupBy("followed").agg(count("followed").alias("followers")).orderBy(desc("followers")).show(5)

"""
Explanation: Performance testing
GZIP Compressed CSV file vs Parquet file
End of explanation
"""


# cache dataframes
socialDF.cache()
socialDFpq.cache()

# remove from cache
#socialDF.unpersist()
#socialDFpq.unpersist()

"""
Explanation: Cached DF vs not cached DF
This time we will cache the 2 previous dataframes (socialDF and socialDFpq) and see how faster is.
End of explanation
"""


%%time
# GZIP Compressed CSV (dataframe cached)
socialDF.groupBy("followed").agg(count("followed").alias("followers")).orderBy(desc("followers")).show(5)

%%time
# Parquet file (dataframe cached)
socialDFpq.groupBy("followed").agg(count("followed").alias("followers")).orderBy(desc("followers")).show(5)

"""
Explanation: Note: The first time we run cached dataframes can be slower, but the next times they should run faster.
End of explanation
"""


