import findspark 
findspark.init()


"""
Explanation: Introduction to PySpark
We'll talk a little bit about the Spark's precursor, Hadoop, and then we'll discuss the advantages and utility of Spark on top of Hadoop.
Next, we'll discuss what Dash offers in conjunction with Spark. 
Finally, we will implement a simple Spark application using the PySpark API. 
In the Beginning, There was Hadoop...

Apache Hadoop - an open source software framework for reliable,scalable, distributed computing on commodity hardware

Main modules:
Hadoop Common - main Hadoop Utilities
HDFS - Hadoop Distributed File System
Hadoop YARN - a task scheduling / cluster resource management module
Hadoop MAP/REDUCE - paralell computing paradigm for big data anaylytics 


Emphasis on reliability and scalability
Pervasive Assumption: node failure is the rule, not the exception 
Task / Cluster management is performed automatically, under the hood


YARN is rack aware 
Tasks are scheduled either on a node where the data is already housed, or preference is given to a node on the same rack 
Reduces overall traffic between racks, increasing throughput 



Big Data Analytics with Hadoop

Store the data using the HDFS 
Define a Mapper class (user defined class extends Mapper) which implements a map function
The map function takes in a pair, &lt;k1,v1&gt; , and maps it to a new value, &lt;k2,v2&gt;
The prototypical example is a word count program : map a document to a set of &lt;k,v&gt;=&lt;w,c&gt; pairs
The document is distributed as chunks throughout the file system
Each chunk of the document becomes a value, v, associated with a key, k, in a &lt;k,v&gt; pair


Split each value into tokens (i.e. - words) associated with an iterator, iter 
For every word in iter, w, create a &lt;k,v&gt;=&lt;w,1&gt; pair 
Output a multi-set of &lt;w,1&gt; pairs for the chunk
We now have one multiset for each chunk of the document


Collect the multisets into a single multiset 




Define a Reducer class which impelements a reduce function 
The reduce function reduces the output from the mapping to a more meaningful state
Reduce the multiset to a set of &lt;w,c&gt; pairs by summing all v with the same k   





A Map/Reduce Example Implementation:
public static class TokenizerMapper
       extends Mapper&lt;Object, Text, Text, IntWritable&gt;{

     private final static IntWritable one = new IntWritable(1);
     private Text word = new Text();

     public void map(Object key, Text value, Context context
                ) throws IOException, InterruptedException {
         StringTokenizer itr = new StringTokenizer(value.toString());
         while (itr.hasMoreTokens()) {
             word.set(itr.nextToken());
             context.write(word, one);
         }
     }
}

public static class IntSumReducer
     extends Reducer&lt;Text,IntWritable,Text,IntWritable&gt; {
     private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable&lt;IntWritable&gt; values, Context context
                ) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
        sum += val.get();
        }
    result.set(sum);
    context.write(key, result);
    }
}


And Lo! the villagers did rejoice!

...for a time...

Some Problems with Hadoop MAP/REDUCE
1. Every application has to be pidgeon-holed into the MAP/REDUCE paradigm
2. Studies showed that up to 90% of computation wall time was spent in file I/O
3. Iterative algorithms were especially slow when implemented with MAP/REDUCE
    * The application may bottle neck due to a low number of particularly slow nodes


Enter Spark


Originally developed by Matei Zaharia and researchers at UC Berkley's AMP Lab
Donated to the Apache Software Foundation in 2013

## RDD - Resilient Distributed Dataset

The main data structure for Spark applications
Fault tolerant multi-set of data partitions available to a computing cluster in shared memory 
Requires cluster management and distributed file system
Supported cluster management includes Spark native, Hadoop YARN, and Apache Mesos
Several supported DFS:
HDFS
MAP-R FDS 
Cassandra 
...


RDDs can be created by any storage source supported by Hadoop



In memory processing and a more diverse API are its main benefits over MAP/REDUCE
Iterative Operations in Hadoop MAP/REDUCE:

versus
Iterative Operations in Spark :

Interactive Operations in Hadoop MAP/REDUCE:

versus
Interactive Operations in Spark:

PySpark : a Python API for programming spark applications
* Originally Spark was written in Scala and most Spark applications were written in either Scala or Java
* Eventually support was extended with APIs for Python and R 
* We are going to work with the Python API : PySpark

The SparkShell
Spark has an REPL interactive shell called the SparkShell

Let's create a simple Spark application...
This piece of the presentation borrows heavily from a July 2017 DataCamp tutorial on machine learning and PySpark. Visit the following URL to see the original tutorial: 
https://www.datacamp.com/community/tutorials/apache-spark-tutorial-machine-learning#gs.Y3MIPIY
Data: California Realestate
We're going to explore some data and run some iterative algorithms on it. You can find the data here: http://www.dcc.fc.up.pt/~ltorgo/Regression/cal_housing.html 
Download the cal_housing.tgz tar ball at the bottom of the page and extract it somewhere obvious.
First we need to use the findspark package inorder to be able to locate the various Spark packages that we need (i.e. - pyspark, etc.) You can download and install findspark using pip as I did, or I'm sure anaconda will also work. There are probably other ways of making sure you can locate the packages you need as well, but this was the simplest and most straight forward I found.
End of explanation
"""


from pyspark.sql import SparkSession

"""
Explanation: NOTE: This is only necessary in the Jupyter Notebook. You should be able to import the necessary packages in a regular Python script without using findspark
The SparkSession is the entry point for any Spark application.
End of explanation
"""


spark = SparkSession.builder\
        .master("local")\
        .appName("LinearRegressionModel")\
        .config("spark.executor.memory","1gb")\
        .getOrCreate()
        
sc = spark.sparkContext

"""
Explanation: Let's create a new SparkSession through the builder attribute and the getOrCreate() method.
End of explanation
"""


rdd = sc.textFile('data/CaliforniaHousing/cal_housing.data')
header = sc.textFile('data/CaliforniaHousing/cal_housing.domain')

"""
Explanation: Initializing the master and appName attributes isn't actually important or critical in this introduction, nor is configuring the memory options for the executor. I've included here for the sake of thoroughness.
NOTE: If you find the Spark tutorial on the Spark documentation web page it includes the following line of code: 
spark = SparkSession.builder().appName(appName).master(master).getOrCreate()

That does not work. You must pass a string as an argument to the appName() and master() methods. 
Moving on...
From here we can create a couple of RDDs: one with the data and another with the domain information, the header
End of explanation
"""


header

"""
Explanation: Part of what allows for some of the speed-up in Spark applications is that Spark evaluations are mostly lazy evals. So executing the following line of code isn't very useful:
End of explanation
"""


header.collect()

"""
Explanation: Instead we have to take an action on the rdd, such as collect() to materialize the data represented by the rdd abstraction
End of explanation
"""


rdd.take(2)

"""
Explanation: NOTE: collect() is a pretty dangerous action: if the RDD is especially large then your executor you may run out of RAM and your application will crash. If you're using especially large data and you just want a peak at it to try to suss out its structure, then try take() or first()
End of explanation
"""


rdd = rdd.map(lambda line: line.split(","))
rdd.take(2)

"""
Explanation: Since we read the data in with textFile() we just have a set of strings separated by commas as our data. Let's split the data into separate entriees using the map() function.
End of explanation
"""


from pyspark.sql import Row

df = rdd.map(lambda line: Row(longitude=line[0], 
                              latitude=line[1], 
                              housingMedianAge=line[2],
                              totalRooms=line[3],
                              totalBedRooms=line[4],
                              population=line[5], 
                              households=line[6],
                              medianIncome=line[7],
                              medianHouseValue=line[8])).toDF()
df.show()

"""
Explanation: Now we have something more closely resembling a collection of records. 
But, notice that the data does not have a header and is mostly unstructured. 
We can fix that by converting the data to a DataFrame. 
I have been told that n general DataFrames perform better than the RDDs, especially when using Python...
End of explanation
"""


df.printSchema()

"""
Explanation: To examine the data types associated with the dataframe printSchemea() method.
End of explanation
"""


from pyspark.sql.types import *

df = df.withColumn("longitude", df["longitude"].cast(FloatType())) \
   .withColumn("latitude", df["latitude"].cast(FloatType())) \
   .withColumn("housingMedianAge",df["housingMedianAge"].cast(FloatType())) \
   .withColumn("totalRooms", df["totalRooms"].cast(FloatType())) \ 
   .withColumn("totalBedRooms", df["totalBedRooms"].cast(FloatType())) \ 
   .withColumn("population", df["population"].cast(FloatType())) \ 
   .withColumn("households", df["households"].cast(FloatType())) \ 
   .withColumn("medianIncome", df["medianIncome"].cast(FloatType())) \ 
   .withColumn("medianHouseValue", df["medianHouseValue"].cast(FloatType()))

"""
Explanation: Since we used the textFile() method to read our data in, the data types are all string
The following would cast all of the columns to floats instead:
End of explanation
"""


from pyspark.sql.types import *

def convertCols(df,names,dataType): #df - a dataframe, names - a list of col names, dataType - the cast conversion type
    for name in names:
        df = df.withColumn(name,df[name].cast(dataType))
    return df

names = ['households', 'housingMedianAge', 'latitude', 'longitude', 'medianHouseValue', 'medianIncome',\
         'population', 'totalBedRooms', 'totalRooms']


df = convertCols(df,names,FloatType())
df.printSchema()

df.show(10)

"""
Explanation: But that seems pretty inefficient, and it is. We can write a function to handle all of this for us:
End of explanation
"""


df.select('population','totalBedrooms').show(10)

"""
Explanation: The pyspark.sql package has lots of convenient data exploration methods built in that support SQL query language execution. For example, we can select by columns:
End of explanation
"""


ndf = df.select('population','totalBedrooms').filter(df['totalBedrooms'] > 500)
ndf.show(10)

"""
Explanation: We can use the filter() method to perform a classic SELECT FROM WHERE query as below:
End of explanation
"""


df.describe().show()

"""
Explanation: And we can get summary statistics pretty easilly too...
End of explanation
"""


# Import all from `sql.functions` 
from pyspark.sql.functions import *

df.show()

# Adjust the values of `medianHouseValue`
df = df.withColumn("medianHouseValue", col("medianHouseValue")/100000)

df.show()


"""
Explanation: Let's do a quick bit of feature engineering and transformation to optimize a linear regression on our feature set...
End of explanation
"""


# Import all from `sql.functions` if you haven't yet
from pyspark.sql.functions import *

# Divide `totalRooms` by `households`
roomsPerHousehold = df.select(col("totalRooms")/col("households"))

# Divide `population` by `households`
populationPerHousehold = df.select(col("population")/col("households"))

# Divide `totalBedRooms` by `totalRooms`
bedroomsPerRoom = df.select(col("totalBedRooms")/col("totalRooms"))

# Add the new columns to `df`
df = df.withColumn("roomsPerHousehold", col("totalRooms")/col("households")) \
   .withColumn("populationPerHousehold", col("population")/col("households")) \
   .withColumn("bedroomsPerRoom", col("totalBedRooms")/col("totalRooms"))
   
# Inspect the result
df.first()

"""
Explanation: We can examine the column of medianHouseValue in the above outputs to make sure that we transformed the data correctly. 
Let's do some more feature engineering and standardization.
End of explanation
"""


# Re-order and select columns
df = df.select("medianHouseValue", 
              "totalBedRooms", 
              "population", 
              "households", 
              "medianIncome", 
              "roomsPerHousehold", 
              "populationPerHousehold", 
              "bedroomsPerRoom")

"""
Explanation: Notice that we're using the col() function to specify that we're using columnar data in our calculations. The col("totalRooms")/col("households") is acting like a numpy array, element wise dividing the results. 
Next we'll use the select() method to reorder the data so that our response variable is
End of explanation
"""


# Import `DenseVector`
from pyspark.ml.linalg import DenseVector

# Define the `input_data` 
input_data = df.rdd.map(lambda x: (x[0], DenseVector(x[1:])))

# Replace `df` with the new DataFrame
df = spark.createDataFrame(input_data, ["label", "features"])

"""
Explanation: Now we're going to actually isolate the response variable of labels from the predictor variables using a DenseVector, which is essentially a numpy ndarray.
End of explanation
"""


# Import `StandardScaler` 
from pyspark.ml.feature import StandardScaler

# Initialize the `standardScaler`
standardScaler = StandardScaler(inputCol="features", outputCol="features_scaled")

# Fit the DataFrame to the scaler
scaler = standardScaler.fit(df)

# Transform the data in `df` with the scaler
scaled_df = scaler.transform(df)

# Inspect the result
scaled_df.take(2)

"""
Explanation: There are all kinds of great machine learning algorithms and functions already built into PySpark in the Spark ML library. If you're interested in more data pipelining, try visiting this page: https://spark.apache.org/docs/latest/ml-pipeline.html
End of explanation
"""


train_data, test_data = scaled_df.randomSplit([.8,.2],seed=1234)

"""
Explanation: We can divide the data into training and testing sets using the PySpark SQL randomSplit() method.
End of explanation
"""


# Import `LinearRegression`
from pyspark.ml.regression import LinearRegression

# Initialize `lr`
lr = LinearRegression(labelCol="label", maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the data to the model
linearModel = lr.fit(train_data)

# Generate predictions
predicted = linearModel.transform(test_data)

# Extract the predictions and the "known" correct labels
predictions = predicted.select("prediction").rdd.map(lambda x: x[0])
labels = predicted.select("label").rdd.map(lambda x: x[0])

# Zip `predictions` and `labels` into a list
predictionAndLabel = predictions.zip(labels).collect()

# Print out first 5 instances of `predictionAndLabel` 
predictionAndLabel[:5]

"""
Explanation: Now we can create the regression model. The original tutorial directs you to the following URL for information on the linear regression model class: 
https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.regression.LinearRegression
End of explanation
"""


# Coefficients for the model
linearModel.coefficients

# Intercept for the model
linearModel.intercept

"""
Explanation: To evaluate the model, we can inspect the model parameters.
End of explanation
"""


# Get the RMSE
linearModel.summary.rootMeanSquaredError

# Get the R2
linearModel.summary.r2

"""
Explanation: And summary data for the model is available as well.
End of explanation
"""


spark.stop()

"""
Explanation: Stop the Spark session...
End of explanation
"""


