<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Spark MLLib
Official documentation
Step1:  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.
Step2:  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
Step3:  Pre-process the data
Step4:  Transform 60 features into MMlib vectors
Step5:  Scale features to have zero mean and unit standard deviation
Step6:  Convert label to numeric index
Step7:  Extract only columns of interest
Step8:  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.
Step9:  Build Model
Step10:  Optimize and fit the model to data
Note that we are looking at optimistic in-sample errors.
Step11:  Post-processing and model evaluation
The GMM is poor at clustering rocks and mines based on the first 2 PC of the sonographic data.
Step12:  Plot discrepancy between predicted and labels
Step13:  Supervised Learning
We will fit a logistic regression model to the data as an example of supervised learning.
Step14:  Using mllib and RDDs
Convert to format expected by regression functions in mllib
Step15:  Split into test and train sets
Step16:  Fit model to training data
Step17:  Evaluate on test data
Step18:  Using the newer ml pipeline
Step19:  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
<ASSISTANT_TASK:>
Python Code:
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
<END_TASK>