<jupyter_start><jupyter_text>The Decision Tree on the Churn Dataset in Spark with Cross Validation<jupyter_code>import sys
sys.path.append("..")
from pyspark.sql import DataFrameReader
from pyspark.sql import SparkSession
from pyspark.ml.feature import IndexToString, Normalizer, StringIndexer, VectorAssembler, VectorIndexer
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml import Pipeline
from helpers.path_translation import translate_to_file_string<jupyter_output><empty_output><jupyter_text>Select the churn file<jupyter_code>inputFile = translate_to_file_string("../data/churn.csv")<jupyter_output><empty_output><jupyter_text>Create the Spark Session<jupyter_code>#create a SparkSession
spark = (SparkSession
       .builder
       .appName("ChurnDecisionTree")
       .getOrCreate())
# create a DataFrame using an ifered Schema 
df = spark.read.option("header", "true") \
       .option("inferSchema", "true") \
       .option("delimiter", ";") \
       .csv(inputFile)<jupyter_output><empty_output><jupyter_text>Data Preparation Transform labels into index<jupyter_code>labelIndexer = StringIndexer().setInputCol("LEAVE").setOutputCol("label").fit(df)
collegeIndexer = StringIndexer().setInputCol("COLLEGE").setOutputCol("COLLEGE_NUM").fit(df)
satIndexer = StringIndexer().setInputCol("REPORTED_SATISFACTION").setOutputCol("REPORTED_SATISFACTION_NUM").fit(df)
usageIndexer = StringIndexer().setInputCol("REPORTED_USAGE_LEVEL").setOutputCol("REPORTED_USAGE_LEVEL_NUM").fit(df)
changeIndexer = StringIndexer().setInputCol("CONSIDERING_CHANGE_OF_PLAN").setOutputCol("CONSIDERING_CHANGE_OF_PLAN_NUM").fit(df)<jupyter_output><empty_output><jupyter_text>Build the feature vector<jupyter_code>featureCols = df.columns.copy()
featureCols.remove("LEAVE")
featureCols.remove("COLLEGE")
featureCols.remove("REPORTED_SATISFACTION")
featureCols.remove("REPORTED_USAGE_LEVEL")
featureCols.remove("CONSIDERING_CHANGE_OF_PLAN")
featureCols = featureCols +["COLLEGE_NUM","REPORTED_SATISFACTION_NUM","REPORTED_USAGE_LEVEL_NUM","CONSIDERING_CHANGE_OF_PLAN_NUM"]<jupyter_output><empty_output><jupyter_text>Build the feature Vector Assembler<jupyter_code>assembler =  VectorAssembler(outputCol="features", inputCols=list(featureCols))<jupyter_output><empty_output><jupyter_text>Build a featureIndexer Automatically identify categorical features, and index them.Features with > 5 distinct values are treated as continuous.<jupyter_code>featureIndexer = VectorIndexer(inputCol="features",outputCol="indexedFeatures", maxCategories=6)<jupyter_output><empty_output><jupyter_text>Convert indexed labels back to original labels<jupyter_code>predConverter = IndexToString(inputCol="prediction",outputCol="predictedLabel",labels=labelIndexer.labels)<jupyter_output><empty_output><jupyter_text>Build the decision tree model<jupyter_code>dt = DecisionTreeClassifier(labelCol="label", featuresCol="features")<jupyter_output><empty_output><jupyter_text>Data Preparation Build a network parameters grid<jupyter_code>paramGrid = ParamGridBuilder().addGrid(dt.maxDepth, [ 5, 10 ]) \
				              .addGrid(dt.minInfoGain, [0.05, 0.025, 0.02, 0.01]) \
				              .addGrid(dt.minInstancesPerNode, [5, 10, 15]) \
                              .addGrid(dt.maxBins, [5, 6, 9]) \
				              .build()<jupyter_output><empty_output><jupyter_text>Split the data<jupyter_code>splits = df.randomSplit([0.6, 0.4 ], 1234)
train = splits[0]
test = splits[1]<jupyter_output><empty_output><jupyter_text>Build a pipeline<jupyter_code>pipeline = Pipeline(stages= [labelIndexer, collegeIndexer, satIndexer,
				usageIndexer, changeIndexer, assembler, featureIndexer,  dt, predConverter])<jupyter_output><empty_output><jupyter_text>Build an evaluator<jupyter_code>evaluator =  BinaryClassificationEvaluator(labelCol="label",rawPredictionCol="rawPrediction", metricName="areaUnderROC")<jupyter_output><empty_output><jupyter_text>Build the Cross Validator<jupyter_code>cv = CrossValidator(estimator=pipeline, evaluator=evaluator,estimatorParamMaps=paramGrid,numFolds=10, parallelism=2)<jupyter_output><empty_output><jupyter_text>Train the Model<jupyter_code>cvModel = cv.fit(train)<jupyter_output><empty_output><jupyter_text>Find out what is the best modelCurrently: - maxBins = 6- minInstancesPerNode = 15- minInfoGain = 0.01- maxDepth = 10<jupyter_code>treeModel = cvModel.bestModel.stages[7]
print("Learned classification tree model:\n",treeModel)
print("Best Params: \n", treeModel.explainParams())<jupyter_output>Learned classification tree model:
 DecisionTreeClassificationModel: uid=DecisionTreeClassifier_597e7a6202f1, depth=10, numNodes=41, numClasses=2, numFeatures=11
Best Params: 
 cacheNodeIds: If false, the algorithm will pass trees to executors to match instances with nodes. If true, the algorithm will cache node IDs for each instance. Caching can speed up training of deeper trees. Users can set how often should the cache be checkpointed or disable it by setting checkpointInterval. (default: False)
checkpointInterval: set checkpoint interval (>= 1) or disable checkpoint (-1). E.g. 10 means that the cache will get checkpointed every 10 iterations. Note: this setting will be ignored if the checkpoint directory is not set in the SparkContext. (default: 10)
featuresCol: features column name. (default: features, current: features)
impurity: Criterion used for information gain calculation (case-insensitive). Supported options: entropy, gini (default: gini)
labelCol: label column name. (defaul[...]<jupyter_text>Test the model<jupyter_code>predictions = cvModel.transform(test)
predictions.select("prediction", "label", "predictedLabel", "LEAVE", "features").show()<jupyter_output>+----------+-----+--------------+-----+--------------------+
|prediction|label|predictedLabel|LEAVE|            features|
+----------+-----+--------------+-----+--------------------+
|       0.0|  1.0|          STAY|LEAVE|[20007.0,36.0,23....|
|       1.0|  1.0|         LEAVE|LEAVE|[20009.0,183.0,18...|
|       1.0|  1.0|         LEAVE|LEAVE|[20022.0,205.0,36...|
|       0.0|  0.0|          STAY| STAY|[20074.0,0.0,47.0...|
|       0.0|  0.0|          STAY| STAY|(11,[0,3,4,6,8],[...|
|       0.0|  1.0|          STAY|LEAVE|[20175.0,191.0,23...|
|       1.0|  1.0|         LEAVE|LEAVE|[20224.0,223.0,11...|
|       1.0|  0.0|         LEAVE| STAY|[20273.0,0.0,30.0...|
|       0.0|  0.0|          STAY| STAY|[20278.0,0.0,69.0...|
|       0.0|  0.0|          STAY| STAY|[20284.0,0.0,5.0,...|
|       1.0|  1.0|         LEAVE|LEAVE|[20288.0,0.0,0.0,...|
|       0.0|  1.0|          STAY|LEAVE|(11,[0,2,3,4,6,9]...|
|       0.0|  1.0|          STAY|LEAVE|[20317.0,85.0,0.0...|
|       1.0|  1.0|      [...]<jupyter_text>Test Error<jupyter_code>accuracy = evaluator.evaluate(predictions)
print("Test Error = " ,(1.0 - accuracy))
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14391882235243675, 'fraction_numerical': 0.0383021294840646, 'mean_word_length': 5.0327586206896555, 'pattern_counts': {'":': 0, '<': 71, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '404883', 'license': 'Apache-2.0', 'n_tokens_mistral': 2568, 'n_tokens_neox': 2392, 'n_words': 416, 'path': 'solutions/churn_decision_tree_cross_validation_solution.ipynb', 'repo_name': 'sturc/bd_and_ds', 'size': '6997'}