from pyspark import SparkContext
sc = SparkContext('local[*]')

from pyspark.sql import SQLContext
sqlc = SQLContext(sc)

"""
Explanation: Spark MLLib

Official documentation: The official documentation is clear, detailed and includes many code examples. You should refer to the official docs for exploration of this rich and rapidly growing library.

MLLib Pipeline
Generally, use of MLLIb for supervised and unsupervised learning follow some or all of the stages in the following template:

Get data
Pre-process the data
Convert data to a form that MLLib functions require (*)
Build a model
Optimize and fit the model to the data
Post-processing and model evaluation

This is often assembled as a pipeline for convenience and reproducibility. This is very similar to what you would do with sklearn, except that MLLib allows you to handle massive datasets by distributing the analysis to multiple computers.
Set up Spark and Spark SQL contexts
End of explanation
"""


from pyspark.ml.feature import VectorAssembler
from pyspark.ml.feature import StandardScaler
from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import PCA
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression

from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.clustering import GaussianMixture
from pyspark.mllib.classification import LogisticRegressionWithLBFGS, LogisticRegressionModel

"""
Explanation: Spark MLLib imports
The older mllib package works on RDDs. The newer ml package works on DataFrames. We will show examples using both, but it is more convenient to use the ml package.
End of explanation
"""


df = (sqlc.read.format('com.databricks.spark.csv')
      .options(header='false', inferschema='true')
      .load('data/sonar.all-data.txt'))

df.printSchema()

"""
Explanation: Unsupervised Learning
We saw this machine learning problem previously with sklearn, where the task is to distinguish rocks from mines using 60 sonar numerical features. We will illustrate some of the mechanics of how to work with MLLib - this is not intended to be a serious attempt at modeling the data.
Obtain data
End of explanation
"""


df = df.withColumnRenamed("C60","label")

"""
Explanation: Pre-process the data
End of explanation
"""


assembler = VectorAssembler(
    inputCols=['C%d' % i for i in range(60)],
    outputCol="features")
output = assembler.transform(df)

"""
Explanation: Transform 60 features into MMlib vectors
End of explanation
"""


standardizer = StandardScaler(withMean=True, withStd=True, 
                              inputCol='features', 
                              outputCol='std_features')
model = standardizer.fit(output)
output = model.transform(output)

"""
Explanation: Scale features to have zero mean and unit standard deviation
End of explanation
"""


indexer = StringIndexer(inputCol="label", outputCol="label_idx")
indexed = indexer.fit(output).transform(output)

"""
Explanation: Convert label to numeric index
End of explanation
"""


sonar = indexed.select(['std_features', 'label', 'label_idx'])

sonar.show(n=3)

"""
Explanation: Extract only columns of interest
End of explanation
"""


pca = PCA(k=2, inputCol="std_features", outputCol="pca")
model = pca.fit(sonar)
transformed = model.transform(sonar)

features = transformed.select('pca').rdd.map(lambda x: np.array(x))

features.take(3)

"""
Explanation: Data conversion
We will first fit a Gaussian Mixture Model with 2 components to the first 2 principal components of the data as an example of unsupervised learning. The GaussianMixture model requires an RDD of vectors, not a DataFrame. Note that pyspark converts numpy arrays to Spark vectors.
End of explanation
"""


gmm = GaussianMixture.train(features, k=2)

"""
Explanation: Build Model
End of explanation
"""


predict = gmm.predict(features).collect()

labels = sonar.select('label_idx').rdd.map(lambda r: r[0]).collect()

"""
Explanation: Optimize and fit the model to data
Note that we are looking at optimistic in-sample errors.
End of explanation
"""


np.corrcoef(predict, labels)

"""
Explanation: Post-processing and model evaluation
The GMM is poor at clustering rocks and mines based on the first 2 PC of the sonographic data.
End of explanation
"""


xs = np.array(features.collect()).squeeze()

fig, axes = plt.subplots(1, 2, figsize=(10, 4))
axes[0].scatter(xs[:, 0], xs[:,1], c=predict)
axes[0].set_title('Predicted')
axes[1].scatter(xs[:, 0], xs[:,1], c=labels)
axes[1].set_title('Labels')
pass

"""
Explanation: Plot discrepancy between predicted and labels
End of explanation
"""


sonar.show(n=3)

"""
Explanation: Supervised Learning
We will fit a logistic regression model to the data as an example of supervised learning.
End of explanation
"""


data = sonar.map(lambda x: LabeledPoint(x[2], x[0]))

"""
Explanation: Using mllib and RDDs
Convert to format expected by regression functions in mllib
End of explanation
"""


train, test = data.randomSplit([0.7, 0.3])

"""
Explanation: Split into test and train sets
End of explanation
"""


model = LogisticRegressionWithLBFGS.train(train)

"""
Explanation: Fit model to training data
End of explanation
"""


y_yhat = test.map(lambda x: (x.label, model.predict(x.features)))
err = y_yhat.filter(lambda x: x[0] != x[1]).count() / float(test.count())
print("Error = " + str(err))

"""
Explanation: Evaluate on test data
End of explanation
"""


transformer = VectorAssembler(inputCols=['C%d' % i for i in range(60)],
                              outputCol="features")
standardizer = StandardScaler(withMean=True, withStd=True, 
                              inputCol='features', 
                              outputCol='std_features')
indexer = StringIndexer(inputCol="C60", outputCol="label_idx")
pca = PCA(k=5, inputCol="std_features", outputCol="pca")
lr = LogisticRegression(featuresCol='std_features', labelCol='label_idx')

pipeline = Pipeline(stages=[transformer, standardizer, indexer, pca, lr])

df = (sqlc.read.format('com.databricks.spark.csv')
      .options(header='false', inferschema='true')
      .load('data/sonar.all-data.txt'))

train, test = df.randomSplit([0.7, 0.3])

model = pipeline.fit(train)

import warnings

with warnings.catch_warnings():
    warnings.simplefilter('ignore')
    prediction = model.transform(test)

score = prediction.select(['label_idx', 'prediction'])
score.show(n=score.count())

acc = score.map(lambda x: x[0] == x[1]).sum() / score.count()
acc

"""
Explanation: Using the newer ml pipeline
End of explanation
"""


from sklearn import svm, grid_search, datasets
from spark_sklearn import GridSearchCV
iris = datasets.load_iris()
parameters = {'kernel':('linear', 'rbf'), 'C':[1, 10]}
svr = svm.SVC()
clf = GridSearchCV(sc, svr, parameters)
clf.fit(iris.data, iris.target)

"""
Explanation: Spark MLLIb and sklearn integration
There is a package that you can install with
bash
pip install spark-sklearn
Basically, it provides the same API as sklearn but uses Spark MLLib under the hood to perform the actual computations in a distributed way (passed in via the SparkContext instance).
Example taken directly from package website
Install if necessary
! pip install spark_sklearn
End of explanation
"""


