<jupyter_start><jupyter_text>Basic Configuration Required installations1. Java    - download from http://www.oracle.com/technetwork/java/javase/downloads/index.html    - install using the downloaded executable file    - install Java to a different path name instead of *Program Files* due to issues relating to the additional whitespace2. Anaconda    - download from https://www.anaconda.com/download/    - install using the downloaded executable file3. Apache Spark     - download from https://spark.apache.org/downloads.html    - unzip to a new folder e.g. *C:\spark*4. Apache Hadoop     - only *winutils.exe* is required    - download from https://github.com/steveloughran/winutils    - create a new folder e.g. *C:\hadoop\bin* where this file should be placed in    - Run command prompt as Adminstrator and execute    ```bash    winutils.exe chmod 777 \tmp\hive    ```5. Findspark    - a utility to locate and initialize pyspark    - install using    ```bash    conda install -c conda-forge findspark    ``` Environment variables- HADOOP_HOME = *path\to\hadoop*- SPARK_HOME = *path\to\spark*- JAVA_HOME = *path\to\JavaJDK*  Import base packages<jupyter_code>import sys, os, shutil
import findspark
# use findspark to locate and initialize pyspark before importing pyspark
findspark.init()
import pyspark<jupyter_output><empty_output><jupyter_text>Check environment<jupyter_code>print("Python Version:", sys.version)
print("Spark Version:", pyspark.__version__)<jupyter_output>Python Version: 3.6.3 |Anaconda, Inc.| (default, Oct 15 2017, 03:27:45) [MSC v.1900 64 bit (AMD64)]
Spark Version: 2.1.1+hadoop2.7<jupyter_text>Example 1: Calculate value of PiAdapted from https://github.com/apache/spark/blob/master/examples/src/main/python/pi.py<jupyter_code>from random import random
from operator import add
from pyspark.sql import SparkSession

def f(_):
    x = random() * 2 - 1
    y = random() * 2 - 1
    return 1 if x ** 2 + y ** 2 <= 1 else 0
    
spark = SparkSession \
        .builder \
        .appName("PythonPi") \
        .getOrCreate()

partitions = 10
num_samples = 10000

count = spark.sparkContext.parallelize(range(1, num_samples + 1), partitions).map(f).reduce(add)

print("Pi is roughly %f" % (4.0 * count / num_samples))

spark.stop()<jupyter_output>Pi is roughly 3.132000<jupyter_text>Example 2A: Perform Binary Classification using Decision TreeAdapted from https://github.com/apache/spark/blob/master/examples/src/main/python/ml/decision_tree_classification_example.py<jupyter_code>from pyspark.ml import Pipeline
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.feature import StringIndexer, VectorIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

from pyspark.sql import SparkSession

spark = SparkSession\
        .builder\
        .appName("DecisionTreeBinaryClassificationExample")\
        .getOrCreate()

data = spark.read.format("libsvm").load("data/sample_libsvm_data.txt")

# Index labels, adding metadata to the label column.
# Fit on whole dataset to include all labels in index.
labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)

# Automatically identify categorical features, and index them.
# We specify maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3], 1234)

# Create a DecisionTree model.
dt = DecisionTreeClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures")

# Chain indexers and tree in a Pipeline
pipeline = Pipeline(stages=[labelIndexer, featureIndexer, dt])

# Train model.  This also runs the indexers.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "indexedLabel", "features").show(5)

# Select (prediction, true label) and compute accuracy
evaluator = MulticlassClassificationEvaluator(
                labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test set accuracy = " + str(accuracy))

treeModel = model.stages[2]

# summary only
print(treeModel.toDebugString)

spark.stop()<jupyter_output>+----------+------------+--------------------+
|prediction|indexedLabel|            features|
+----------+------------+--------------------+
|       1.0|         1.0|(692,[95,96,97,12...|
|       1.0|         1.0|(692,[98,99,100,1...|
|       1.0|         1.0|(692,[100,101,102...|
|       1.0|         1.0|(692,[124,125,126...|
|       1.0|         1.0|(692,[127,128,129...|
+----------+------------+--------------------+
only showing top 5 rows

Test set accuracy = 0.9642857142857143
DecisionTreeClassificationModel (uid=DecisionTreeClassifier_4141bc281b063407d90d) of depth 1 with 3 nodes
  If (feature 434 <= 0.0)
   Predict: 1.0
  Else (feature 434 > 0.0)
   Predict: 0.0<jupyter_text>Example 2B: Perform Multiclass Classification using Decision Tree<jupyter_code>from pyspark.ml import Pipeline
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.feature import StringIndexer, VectorIndexer
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

from pyspark.sql import SparkSession

spark = SparkSession\
        .builder\
        .appName("DecisionTreeMulticlassClassificationExample")\
        .getOrCreate()

data = spark.read.format("libsvm").load("data/sample_multiclass_classification_data.txt")

# Index labels, adding metadata to the label column.
# Fit on whole dataset to include all labels in index.
labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)

# Automatically identify categorical features, and index them.
# We specify maxCategories so features with > 4 distinct values are treated as continuous.
featureIndexer =\
    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)

# Split the data into training and test sets (40% held out for testing)
(trainingData, testData) = data.randomSplit([0.6, 0.4], 1234)

# Create a DecisionTree model.
dt = DecisionTreeClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures")

# Chain indexers and tree in a Pipeline
pipeline = Pipeline(stages=[labelIndexer, featureIndexer, dt])

# Train model.  This also runs the indexers.
model = pipeline.fit(trainingData)

# Make predictions.
predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "indexedLabel", "features").show(5)

# Select (prediction, true label) and compute accuracy
evaluator = MulticlassClassificationEvaluator(
                labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test set accuracy = " + str(accuracy))

treeModel = model.stages[2]

# summary only
print(treeModel.toDebugString)

spark.stop()<jupyter_output>+----------+------------+--------------------+
|prediction|indexedLabel|            features|
+----------+------------+--------------------+
|       2.0|         1.0|(4,[0,1,2,3],[-0....|
|       2.0|         1.0|(4,[0,1,2,3],[-0....|
|       1.0|         1.0|(4,[0,1,2,3],[-0....|
|       2.0|         1.0|(4,[0,1,2,3],[-0....|
|       1.0|         1.0|(4,[0,1,2,3],[-0....|
+----------+------------+--------------------+
only showing top 5 rows

Test set accuracy = 0.9215686274509803
DecisionTreeClassificationModel (uid=DecisionTreeClassifier_4ef0987fa2e49c6e229a) of depth 4 with 13 nodes
  If (feature 2 <= -0.694915)
   Predict: 0.0
  Else (feature 2 > -0.694915)
   If (feature 2 <= 0.322034)
    If (feature 3 <= 0.25)
     Predict: 2.0
    Else (feature 3 > 0.25)
     If (feature 0 <= -0.111111)
      Predict: 2.0
     Else (feature 0 > -0.111111)
      Predict: 1.0
   Else (feature 2 > 0.322034)
    If (feature 3 <= 0.25)
     If (feature 0 <= -0.0555556)
      Predict: 2.0
     Else [...]<jupyter_text>Example 3: Perform Multiclass Classification using Multilayer PerceptronAdapted from https://github.com/apache/spark/blob/master/examples/src/main/python/ml/multilayer_perceptron_classification.py<jupyter_code>from pyspark.ml.classification import MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

from pyspark.sql import SparkSession

spark = SparkSession\
        .builder\
        .appName("multilayer_perceptron_classification_example")\
        .getOrCreate()
        
 # Load training data
data = spark.read.format("libsvm").load("data/sample_multiclass_classification_data.txt")

# Split the data into train and test
splits = data.randomSplit([0.6, 0.4], 1234)
train = splits[0]
test = splits[1]

# specify layers for the neural network:
# input layer of size 4 (features), two intermediate of size 5 and 4
# and output of size 3 (classes)
layers = [4, 5, 4, 3]

# create the trainer and set its parameters
trainer = MultilayerPerceptronClassifier(maxIter=100, layers=layers, blockSize=128, seed=1234)

# train the model
model = trainer.fit(train)

# compute accuracy on the test set
predictions = model.transform(test)

# Select example rows to display.
predictions.select("prediction", "label", "features").show(5)

# Select (prediction, true label) and compute accuracy
evaluator = MulticlassClassificationEvaluator(
                labelCol="label", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test set accuracy = " + str(accuracy))

spark.stop()<jupyter_output>+----------+-----+--------------------+
|prediction|label|            features|
+----------+-----+--------------------+
|       2.0|  0.0|(4,[0,1,2,3],[-0....|
|       0.0|  0.0|(4,[0,1,2,3],[-0....|
|       0.0|  0.0|(4,[0,1,2,3],[-0....|
|       2.0|  0.0|(4,[0,1,2,3],[-0....|
|       2.0|  0.0|(4,[0,1,2,3],[-0....|
+----------+-----+--------------------+
only showing top 5 rows

Test set accuracy = 0.9019607843137255<jupyter_text>Example 4: End-to-End Workflow using Decision TreeUsing mushroom dataset from https://archive.ics.uci.edu/ml/datasets/MushroomThis example describes the basic workflow consisting of the following phases:1. Prepare Spark Environment2. Prepare Data    1. Read Data    2. Review Data    3. Transform Data        1. Transform Label        2. Transform Features        3. Assemble Features3. Create Train-Test Split4. Create Machine Learning Object5. Create Pipeline6. Train Machine Learning Object7. Evaluate Trained Machine Learning Object    1. Perform Predictions    2. Compute Accuracy    3. Display Trained Model8. Stop Spark Session  1. Prepare Spark Environment  Start Spark Sessionusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.sql.htmlmodule-pyspark.sql<jupyter_code>from pyspark.ml import Pipeline
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.feature import StringIndexer, VectorIndexer, VectorAssembler
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

from pyspark.sql import SparkSession

spark = SparkSession\
        .builder\
        .appName("DecisionTreeBinaryClassificationExampleOnMushrooms")\
        .getOrCreate()<jupyter_output><empty_output><jupyter_text>2. Prepare Data  2.A. Read Data- data format: csv- header row: true- infer schema automatically: trueusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.sql.htmlmodule-pyspark.sql<jupyter_code># returns a DataFrame
df = spark.read.format("csv")\
            .option("header", "true")\
            .option("inferSchema", "true")\
            .load("data/mushrooms.csv")<jupyter_output><empty_output><jupyter_text>2.B. Review Datausing https://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.htmlpyspark.sql.DataFrame<jupyter_code>print("Number of data rows:", df.count())
df.printSchema() # or data.dtypes or df.columns
df.describe("class", "cap-shape", "cap-surface", "cap-color").show()
df.first() # or df.head(1) or df.show(1)<jupyter_output><empty_output><jupyter_text>2.C. Transform DataThis is a three-step process:    (a) Transform label     (b) Transform features     (c) Assemble features- pyspark.ml.feature.StringIndexer: a label indexer that maps a string column of labels to an ML column of label indices.- pyspark.ml.feature.VectorIndexer: for indexing categorical feature columns in a dataset of Vector.- pyspark.ml.feature.VectorAssembler: a transformer that combines a given list of columns into a single vector columnusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.ml.htmlmodule-pyspark.ml.feature  2.C.a. Transform LabelUsing StringIndexer to index labels, adding metadata to the label column<jupyter_code>labelIndexer = StringIndexer(inputCol="class", outputCol="indexedLabel")<jupyter_output><empty_output><jupyter_text>2.C.b. Transform FeaturesUsing StringIndexer again because VectorIndexer does not allow strings<jupyter_code>categorical_columns = df.columns[1:]

featureIndexers = [StringIndexer(inputCol=col, outputCol='stringindexed_' + col) for col in categorical_columns]<jupyter_output><empty_output><jupyter_text>2.C.c. Assemble FeaturesUsing VectorAssembler to combine features in a single vector column<jupyter_code>inputFeatures = ['stringindexed_' + col for col in categorical_columns]

assembler = VectorAssembler(
                inputCols=inputFeatures,
                outputCol="indexedFeatures")<jupyter_output><empty_output><jupyter_text>3. Create Train-Test Split- Split the data into training and test sets (30% held out for testing)- Use seed=1234 for reproducibilityusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.sql.htmlpyspark.sql.DataFrame.randomSplit<jupyter_code>(trainingData, testData) = df.randomSplit([0.7, 0.3], 1234)<jupyter_output><empty_output><jupyter_text>4. Create Machine Learning Object- Create a Decision Tree modelusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.ml.htmlpyspark.ml.classification.DecisionTreeClassifier<jupyter_code>clf = DecisionTreeClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures")<jupyter_output><empty_output><jupyter_text>5. Create PipelineChain indexers and model in a Pipeline1. labelIndexer2. featureIndexers3. assembler4. clfusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.ml.htmlpyspark.ml.Pipeline<jupyter_code>pipeline = Pipeline(stages=[labelIndexer] + featureIndexers + [assembler, clf])<jupyter_output><empty_output><jupyter_text>6. Train Machine Learning Object- Execute the stages in the pipeline including featurization and model training on test data using fitusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.ml.htmlpyspark.ml.Pipeline.fit<jupyter_code>model = pipeline.fit(trainingData)<jupyter_output><empty_output><jupyter_text>7. Evaluate Trained Machine Learning Object(1) Perform predictions against the test data by using transformusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.ml.htmlpyspark.ml.PipelineModel.transform<jupyter_code>predictions = model.transform(testData)

# Select example rows to display.
predictions.select("prediction", "indexedLabel", "indexedFeatures").show(5)<jupyter_output>+----------+------------+--------------------+
|prediction|indexedLabel|     indexedFeatures|
+----------+------------+--------------------+
|       0.0|         0.0|(22,[0,1,2,6,8,9,...|
|       0.0|         0.0|(22,[0,1,2,6,8,9,...|
|       0.0|         0.0|(22,[0,1,2,6,8,9,...|
|       0.0|         0.0|(22,[0,1,2,6,8,9,...|
|       0.0|         0.0|(22,[0,1,2,6,8,9,...|
+----------+------------+--------------------+
only showing top 5 rows<jupyter_text>(2) Compute accuracy against the test data using Evaluatorusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.ml.htmlpyspark.ml.evaluation.MulticlassClassificationEvaluator<jupyter_code>evaluator = MulticlassClassificationEvaluator(
                labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test set accuracy = " + str(accuracy))<jupyter_output>Test set accuracy = 0.9987557030277893<jupyter_text>(3) Display trained decision tree using toDebugStringusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.ml.htmlpyspark.ml.classification.DecisionTreeClassificationModel.toDebugString<jupyter_code>treeModel = model.stages[-1] # last stage in Pipeline
# summary only
print(treeModel.toDebugString)<jupyter_output>DecisionTreeClassificationModel (uid=DecisionTreeClassifier_4a6caa8c22c7ca2acac7) of depth 5 with 13 nodes
  If (feature 4 in {0.0,4.0,5.0,8.0})
   If (feature 19 in {0.0,1.0,2.0,3.0,5.0,6.0,7.0,8.0})
    If (feature 14 in {0.0,1.0,2.0,3.0,5.0,6.0})
     If (feature 14 in {0.0,1.0,2.0,5.0,6.0})
      If (feature 21 in {0.0,1.0,2.0,4.0,5.0,6.0})
       Predict: 0.0
      Else (feature 21 not in {0.0,1.0,2.0,4.0,5.0,6.0})
       Predict: 0.0
     Else (feature 14 not in {0.0,1.0,2.0,5.0,6.0})
      If (feature 10 in {0.0})
       Predict: 0.0
      Else (feature 10 not in {0.0})
       Predict: 1.0
    Else (feature 14 not in {0.0,1.0,2.0,3.0,5.0,6.0})
     Predict: 1.0
   Else (feature 19 not in {0.0,1.0,2.0,3.0,5.0,6.0,7.0,8.0})
    Predict: 1.0
  Else (feature 4 not in {0.0,4.0,5.0,8.0})
   Predict: 1.0<jupyter_text>8. Stop Spark Sessionusing https://spark.apache.org/docs/2.1.1/api/python/pyspark.sql.htmlpyspark.sql.SparkSession.stop<jupyter_code>spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1495704613984184, 'fraction_numerical': 0.04665187460886386, 'mean_word_length': 3.7649769585253456, 'pattern_counts': {'":': 0, '<': 81, '<?xml version=': 0, '>': 80, 'https://': 19, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 9, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '20', 'dir': 'jupyter-structured-clean-dedup', 'id': '133017', 'license': 'MIT', 'n_tokens_mistral': 6176, 'n_tokens_neox': 5689, 'n_words': 1328, 'path': 'Getting Started with PySpark on Windows 10.ipynb', 'repo_name': 'ahfant/spark', 'size': '17577'}