<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Decision Tree Classification
Create entry points to spark
Step1:  Decision tree classification with pyspark
Import data
Step2:  Process categorical columns
The following code does three things with pipeline
Step3:  Build StringIndexer stages
Step4:  Build OneHotEncoder stages
Step5:  Build VectorAssembler stage
Step6:  Build pipeline model
Step7:  Fit pipeline model
Step8:  Transform data
Step9:  Split data into training and test datasets
Step10:  Build cross-validation model
Estimator
Step11:  Parameter grid
Step12:  Evaluator
Step13:  Build cross-validation model
Step14:  Fit cross-validation mode
Step15:  Prediction
Step16:  Prediction on training data
Step17:  Prediction on test data
Step18:  Confusion matrix
Pyspark doesn’t have a function to calculate the confusion matrix automatically, but we can still easily get a confusion matrix with a combination use of several methods from the RDD class.
Step19:  Parameters from the best model
<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: Decision Tree Classification
Create entry points to spark
End of explanation
cuse = spark.read.csv('data/cuse_binary.csv', header=True, inferSchema=True)
cuse.show(5)
Explanation: Decision tree classification with pyspark
Import data
End of explanation
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml import Pipeline
# categorical columns
categorical_columns = cuse.columns[0:3]
Explanation: Process categorical columns
The following code does three things with pipeline:
StringIndexer all categorical columns
OneHotEncoder all categorical index columns
VectorAssembler all feature columns into one vector column
Categorical columns
End of explanation
stringindexer_stages = [StringIndexer(inputCol=c, outputCol='strindexed_' + c) for c in categorical_columns]
# encode label column and add it to stringindexer_stages
stringindexer_stages += [StringIndexer(inputCol='y', outputCol='label')]
Explanation: Build StringIndexer stages
End of explanation
onehotencoder_stages = [OneHotEncoder(inputCol='strindexed_' + c, outputCol='onehot_' + c) for c in categorical_columns]
Explanation: Build OneHotEncoder stages
End of explanation
feature_columns = ['onehot_' + c for c in categorical_columns]
vectorassembler_stage = VectorAssembler(inputCols=feature_columns, outputCol='features') 
Explanation: Build VectorAssembler stage
End of explanation
# all stages
all_stages = stringindexer_stages + onehotencoder_stages + [vectorassembler_stage]
pipeline = Pipeline(stages=all_stages)
Explanation: Build pipeline model
End of explanation
pipeline_model = pipeline.fit(cuse)
Explanation: Fit pipeline model
End of explanation
final_columns = feature_columns + ['features', 'label']
cuse_df = pipeline_model.transform(cuse).\
            select(final_columns)
            
cuse_df.show(5)
Explanation: Transform data
End of explanation
training, test = cuse_df.randomSplit([0.8, 0.2], seed=1234)
Explanation: Split data into training and test datasets
End of explanation
from pyspark.ml.regression import GeneralizedLinearRegression
from pyspark.ml.classification import LogisticRegression, DecisionTreeClassifier
dt = DecisionTreeClassifier(featuresCol='features', labelCol='label')
Explanation: Build cross-validation model
Estimator
End of explanation
from pyspark.ml.tuning import ParamGridBuilder
param_grid = ParamGridBuilder().\
    addGrid(dt.maxDepth, [2,3,4,5]).\
    build()
Explanation: Parameter grid
End of explanation
from pyspark.ml.evaluation import BinaryClassificationEvaluator
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", metricName="areaUnderROC")
Explanation: Evaluator
End of explanation
from pyspark.ml.tuning import CrossValidator
cv = CrossValidator(estimator=dt, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=4)
Explanation: Build cross-validation model
End of explanation
cv_model = cv.fit(cuse_df)
Explanation: Fit cross-validation mode
End of explanation
show_columns = ['features', 'label', 'prediction', 'rawPrediction', 'probability']
Explanation: Prediction
End of explanation
pred_training_cv = cv_model.transform(training)
pred_training_cv.select(show_columns).show(5, truncate=False)
Explanation: Prediction on training data
End of explanation
pred_test_cv = cv_model.transform(test)
pred_test_cv.select(show_columns).show(5, truncate=False)
Explanation: Prediction on test data
End of explanation
label_and_pred = cv_model.transform(cuse_df).select('label', 'prediction')
label_and_pred.rdd.zipWithIndex().countByKey()
Explanation: Confusion matrix
Pyspark doesn’t have a function to calculate the confusion matrix automatically, but we can still easily get a confusion matrix with a combination use of several methods from the RDD class.
End of explanation
print('The best MaxDepth is:', cv_model.bestModel._java_obj.getMaxDepth())
Explanation: Parameters from the best model
End of explanation
<END_TASK>