from pyspark.sql import SparkSession

"""
Explanation: Introduction to Spark SQL via PySpark

Goals:

Get familiarized with the basics of Spark SQL and PySpark
Learn to create a SparkSession
Verify if Jupyter can talk to Spark Master

References:
* https://spark.apache.org/docs/latest/api/python/pyspark.html
* https://spark.apache.org/docs/latest/sql-getting-started.html
* https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.SparkSession
* https://jaceklaskowski.gitbooks.io/mastering-spark-sql/
* http://people.csail.mit.edu/matei/papers/2015/sigmod_spark_sql.pdf
What is Spark SQL?

It is a Spark module that leverages Sparks functional programming APIs to allow SQL relational processing tasks on (semi)structured data.
Spark SQL provides Spark with more information about the structure of both the data and the computation being performed

What is PySpark?
PySpark is the Python API for Spark.
How do I start using Pyspark and SparkSQL?

Start by importing the PySpark SQL SparkSession Class and creating a SparkSession instance .
A SparkSession class is considered the entry point to programming Spark with the Dataset and DataFrame API.
A SparkSession can be used create DataFrames, register DataFrames as tables, execute SQL over tables, cache tables, and read parquet files.

Import SparkSession Class
End of explanation
"""


spark = SparkSession.builder \
    .appName("Python Spark SQL basic example") \
    .master("spark://helk-spark-master:7077") \
    .enableHiveSupport() \
    .getOrCreate()

"""
Explanation: What is a SparkSession?

It is the driver process that controls a spark application
A SparkSession instance is responsible for executing the driver program’s commands (code) across executors (in a cluster) to complete a given task.
You can have as many SparkSessions as you want in a single Spark application.

How do I create a SparkSession?

You can use the SparkSession class attribute called Builder.
The class attribute builder allows you to run some of the following functions:
appName: Sets a name for the application
master: URL for the Spark master (Local or Spark standalone cluster)
enableHiveSupport: Enables Hive support, including connectivity to a persistent Hive metastore, support for Hive serdes, and Hive user-defined functions.
getOrCreate:Gets an existing SparkSession or, if there is no existing one, creates a new one based on the options set in this builder.



Create a SparkSession instance

Define a spark variable
Pass values to the appName and master functions
For the master function, we are going to use the HELK's Spark Master container (helk-spark-master)
End of explanation
"""


spark

"""
Explanation: Check the SparkSession variable
End of explanation
"""


first_df = spark.range(10).toDF("numbers")

first_df.show()

"""
Explanation: What is a Dataframe?

In Spark, a dataframe is the most common Structured API, and it is used to represent data in a table format with rows and columns.
Think of a dataframe as a spreadsheet with headers. The difference is that one Spark Dataframe can be distributed across several computers due to its large size or high computation requirements for faster analysis.
The list of column names from a dataframe with its respective data types is called the schema

Is a Spark Dataframe the same as a Python Pandas Dataframe?

A Python dataframe sits on one computer whereas a Spark Dataframe, once again, can be distributed across several computers.
PySpark allows the conversion from Python Pandas dataframes to Spark dataframes. 

Create your first Dataframe
Let's create our first dataframe by using range and toDF functions.
* One column named numbers
* 10 rows containing numbers from 0-9
range(start, end=None, step=1, numPartitions=None)
* Create a DataFrame with single pyspark.sql.types.LongType column named id, containing elements in a range from start to end (exclusive) with step value step.
toDF(*cols)
* Returns a new class:DataFrame that with new specified column names
End of explanation
"""


dog_data=[['Pedro','Doberman',3],['Clementine','Golden Retriever',8],['Norah','Great Dane',6]\
         ,['Mabel','Austrailian Shepherd',1],['Bear','Maltese',4],['Bill','Great Dane',10]]
dog_df=spark.createDataFrame(dog_data, ['name','breed','age'])

dog_df.show()

"""
Explanation: Create another Dataframe
createDataFrame(data, schema=None, samplingRatio=None, verifySchema=True)

Creates a DataFrame from an RDD, a list or a pandas.DataFrame.
When schema is a list of column names, the type of each column will be inferred from data.
When schema is None, it will try to infer the schema (column names and types) from data, which should be an RDD of Row, or namedtuple, or dict.
End of explanation
"""


dog_df.schema

"""
Explanation: Check the Dataframe schema

We are going to do apply a concept called schema inference which lets spark takes its best guess at figuring out the schema.
Spark reads part of the dataframe and then tries to parse the types of data in each row. 
You can also define a strict schema when you read in data which does not let Spark guess. This is recommended for production use cases. 

schema
* Returns the schema of this DataFrame as a pyspark.sql.types.StructType.
End of explanation
"""


dog_df.printSchema()

"""
Explanation: printSchema()
* Prints out the schema in the tree format
End of explanation
"""


dog_df.select("name").show()

"""
Explanation: Access Dataframe Columns
select(*cols)
* Projects a set of expressions and returns a new DataFrame.
Access Dataframes's columns by attribute (df.name):
End of explanation
"""


dog_df.select(dog_df["name"]).show()

"""
Explanation: Access Dataframe's columns by indexing (df['name']). 
* According to Sparks documentation, the indexing form is the recommended one because it is future proof and won’t break with column names that are also attributes on the DataFrame class.
End of explanation
"""


dog_df.filter(dog_df["age"] > 4).show()

"""
Explanation: Filter Dataframe
filter(condition)
* Filters rows using the given condition.
Select dogs that are older than 4 years
End of explanation
"""


dog_df.groupBy(dog_df["age"]).count().show()

"""
Explanation: Group Dataframe
groupBy(*cols)
* Groups the DataFrame using the specified columns, so we can run aggregation on them. See GroupedData for all the available aggregate functions.
group dogs and count them by their age
End of explanation
"""


dog_df.createOrReplaceTempView("dogs")

sql_dog_df = spark.sql("SELECT * FROM dogs")
sql_dog_df.show()

sql_dog_df = spark.sql("SELECT * FROM dogs WHERE name='Pedro'")
sql_dog_df.show()

"""
Explanation: Run SQL queries on your Dataframe
createOrReplaceTempView(name)
* Creates or replaces a local temporary view with this DataFrame.
* The lifetime of this temporary table is tied to the SparkSession that was used to create this DataFrame.
Register the current Dataframe as a SQL temporary view
End of explanation
"""


