<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Create entry points to spark
Step1:  load iris data
Step2:  Merge features to create a features column
Step3:  Index label column with StringIndexer
Import libraries
Step4:  Build pipeline
Try to use pipeline whenever you can to get used to this format.
Step5:  Transform data
Step6:  Check the data one more time
Step7:  Naive Bayes classification
Split data into training and test sets
Step8:  Build cross-validation model
Estimator
Step9:  Parameter grid
Step10:  Evaluator
There are three categories in the label column. Therefore, we use MulticlassClassificationEvaluator
Step11:  Build cross-validation model
Step12:  Fit cross-validation model
Step13:  Prediction on training and test sets
Step14:  Best model from cross validation
Step15:  Prediction accurary
Four accuracy matrices are avaiable for this evaluator. 
* f1
* weightedPrecision
* weightedRecall
* accuracy
Prediction accuracy on training data
Step16:  Prediction accuracy on test data
Step17:  Confusion matrix
Confusion matrix on training data
Step18:  Confusion matrix on test data
<ASSISTANT_TASK:>
Python Code:
from pyspark import SparkContext
sc = SparkContext(master = 'local')
from pyspark.sql import SparkSession
spark = SparkSession.builder \
          .appName("Python Spark SQL basic example") \
          .config("spark.some.config.option", "some-value") \
          .getOrCreate()
Explanation: Create entry points to spark
End of explanation
iris = spark.read.csv('data/iris.csv', header=True, inferSchema=True)
iris.show(5)
iris.dtypes
iris.describe().show()
Explanation: load iris data
End of explanation
from pyspark.ml.linalg import Vectors
from pyspark.sql import Row
iris2 = iris.rdd.map(lambda x: Row(features=Vectors.dense(x[:-1]), species=x[-1])).toDF()
iris2.show(5)
Explanation: Merge features to create a features column
End of explanation
from pyspark.ml.feature import StringIndexer
from pyspark.ml import Pipeline
Explanation: Index label column with StringIndexer
Import libraries
End of explanation
stringindexer = StringIndexer(inputCol='species', outputCol='label')
stages = [stringindexer]
pipeline = Pipeline(stages=stages)
Explanation: Build pipeline
Try to use pipeline whenever you can to get used to this format.
End of explanation
iris_df = pipeline.fit(iris2).transform(iris2)
iris_df.show(5)
Explanation: Transform data
End of explanation
iris_df.describe().show(5)
iris_df.dtypes
Explanation: Check the data one more time
End of explanation
train, test = iris_df.randomSplit([0.8, 0.2], seed=1234)
Explanation: Naive Bayes classification
Split data into training and test sets
End of explanation
from pyspark.ml.classification import NaiveBayes
naivebayes = NaiveBayes(featuresCol="features", labelCol="label")
Explanation: Build cross-validation model
Estimator
End of explanation
from pyspark.ml.tuning import ParamGridBuilder
param_grid = ParamGridBuilder().\
    addGrid(naivebayes.smoothing, [0, 1, 2, 4, 8]).\
    build()
Explanation: Parameter grid
End of explanation
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator = MulticlassClassificationEvaluator()
Explanation: Evaluator
There are three categories in the label column. Therefore, we use MulticlassClassificationEvaluator
End of explanation
from pyspark.ml.tuning import CrossValidator
crossvalidator = CrossValidator(estimator=naivebayes, estimatorParamMaps=param_grid, evaluator=evaluator)
Explanation: Build cross-validation model
End of explanation
crossvalidation_mode = crossvalidator.fit(train)
Explanation: Fit cross-validation model
End of explanation
pred_train = crossvalidation_mode.transform(train)
pred_train.show(5)
pred_test = crossvalidation_mode.transform(test)
pred_test.show(5)
Explanation: Prediction on training and test sets
End of explanation
print("The parameter smoothing has best value:",
      crossvalidation_mode.bestModel._java_obj.getSmoothing())
Explanation: Best model from cross validation
End of explanation
print('training data (f1):', evaluator.setMetricName('f1').evaluate(pred_train), "\n",
     'training data (weightedPrecision): ', evaluator.setMetricName('weightedPrecision').evaluate(pred_train),"\n",
     'training data (weightedRecall): ', evaluator.setMetricName('weightedRecall').evaluate(pred_train),"\n",
     'training data (accuracy): ', evaluator.setMetricName('accuracy').evaluate(pred_train))
Explanation: Prediction accurary
Four accuracy matrices are avaiable for this evaluator. 
* f1
* weightedPrecision
* weightedRecall
* accuracy
Prediction accuracy on training data
End of explanation
print('test data (f1):', evaluator.setMetricName('f1').evaluate(pred_test), "\n",
     'test data (weightedPrecision): ', evaluator.setMetricName('weightedPrecision').evaluate(pred_test),"\n",
     'test data (weightedRecall): ', evaluator.setMetricName('weightedRecall').evaluate(pred_test),"\n",
     'test data (accuracy): ', evaluator.setMetricName('accuracy').evaluate(pred_test))
Explanation: Prediction accuracy on test data
End of explanation
train_conf_mat = pred_train.select('label', 'prediction')
train_conf_mat.rdd.zipWithIndex().countByKey()
Explanation: Confusion matrix
Confusion matrix on training data
End of explanation
test_conf_mat = pred_test.select('label', 'prediction')
test_conf_mat.rdd.zipWithIndex().countByKey()
Explanation: Confusion matrix on test data
End of explanation
<END_TASK>