<jupyter_start><jupyter_text>Churn Dataset Correlation Calculation in Spark<jupyter_code>import sys
sys.path.append("..")
from helpers.path_translation import translate_to_file_string
from helpers.data_prep_and_print import print_df
from pyspark.ml.feature import IndexToString, Normalizer, StringIndexer, VectorAssembler, VectorIndexer
from pyspark.ml.stat import Correlation, ChiSquareTest

from pyspark.sql import DataFrameReader
from pyspark.sql import SparkSession<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("Churn Proprocessing")
       .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>Do the Data Preparation<jupyter_code>labeledData = labelIndexer.transform(df)
print(labeledData.printSchema())
indexedLabedData = collegeIndexer.transform(satIndexer.transform(usageIndexer.transform(changeIndexer.transform(labeledData))))
labeledPointData = assembler.transform(indexedLabedData)
print_df(labeledPointData,10)
# TODO Calc correlation matrix and pretty print it
corr_matrix = Correlation.corr(labeledPointData, "features")
print (str(corr_matrix.head()))
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11373390557939914, 'fraction_numerical': 0.000715307582260372, 'mean_word_length': 10.278225806451612, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 29, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '171593', 'license': 'Apache-2.0', 'n_tokens_mistral': 965, 'n_tokens_neox': 929, 'n_words': 122, 'path': 'exercises/churn_correlation.ipynb', 'repo_name': 'sturc/bd_and_ds', 'size': '2796'}