<jupyter_start><jupyter_text>Getting started with MLlib - binary classification example This tutorial is designed to get you started with Apache Spark MLlib. It investigates a binary classification problem - can you predict if an individual's income is greater than $50,000 based on demographic data? The dataset is from the [UCI Machine Learning Repository](https://archive.ics.uci.edu/ml/datasets/Adult) and is provided with Databricks Runtime. This notebook demonstrates some of the capabilities available in MLlib, including tools for data preprocessing, machine learning pipelines, and several different machine learning algorithms.This notebook includes the following steps:0. Load the dataset0. Feature preprocessing0. Define the model0. Build the pipeline0. Evaluate the model0. Hyperparameter tuning0. Make predictions and evaluate model performance RequirementsDatabricks Runtime 7.0 or above or Databricks Runtime 7.0 ML or above. If you are running Databricks Runtime 6.x or Databricks Runtime 6.x ML, see ([AWS](https://docs.databricks.com/getting-started/spark/machine-learning.html)|[Azure](https://docs.microsoft.com/azure/databricks/getting-started/spark/machine-learning/)) for the correct notebook.  Step 1. Load the dataset Use Databricks utilities to view the first few rows of the data.<jupyter_code>%fs head --maxBytes=1024 databricks-datasets/adult/adult.data<jupyter_output><empty_output><jupyter_text>Because the dataset does not include column names, create a schema to assign column names and datatypes.<jupyter_code>schema = """`age` DOUBLE,
`workclass` STRING,
`fnlwgt` DOUBLE,
`education` STRING,
`education_num` DOUBLE,
`marital_status` STRING,
`occupation` STRING,
`relationship` STRING,
`race` STRING,
`sex` STRING,
`capital_gain` DOUBLE,
`capital_loss` DOUBLE,
`hours_per_week` DOUBLE,
`native_country` STRING,
`income` STRING"""

dataset = spark.read.csv("/databricks-datasets/adult/adult.data", schema=schema)
display(dataset)<jupyter_output><empty_output><jupyter_text>Randomly split data into training and test sets, and set seed for reproducibility.It's best to split the data before doing any preprocessing. This allows the test dataset to more closely simulate new data when we evaluate the model.<jupyter_code>trainDF, testDF = dataset.randomSplit([0.8, 0.2], seed=42)
print(trainDF.cache().count())  # Cache because accessing training data multiple times
print(testDF.count())<jupyter_output><empty_output><jupyter_text>Let's review the data.<jupyter_code>display(trainDF)<jupyter_output><empty_output><jupyter_text>What's the distribution of the number of `hours_per_week`?<jupyter_code>display(trainDF.select("hours_per_week").summary())<jupyter_output><empty_output><jupyter_text>How about `education` status?<jupyter_code>display(trainDF.groupBy("education").count().sort("count", ascending=False))<jupyter_output><empty_output><jupyter_text>Background: Transformers, estimators, and pipelinesThree important concepts in MLlib machine learning that are illustrated in this notebook are **Transformers**, **Estimators**, and **Pipelines**. - **Transformer**: Takes a DataFrame as input, and returns a new DataFrame. Transformers do not learn any parameters from the data and simply apply rule-based transformations to either prepare data for model training or generate predictions using a trained MLlib model. You call a transformer with a `.transform()` method.- **Estimator**: Learns (or "fits") parameters from your DataFrame via a `.fit()` method and returns a Model, which is a transformer.- **Pipeline**: Combines multiple steps into a single workflow that can be easily run. Creating a machine learning model typically involves setting up many different steps and iterating over them. Pipelines help you automate this process.For more information:[Estimators, transformers, and pipelines](https://spark.apache.org/docs/latest/ml-guide.htmloverview-estimators-transformers-and-pipelines-sparkml)  Step 2. Feature preprocessing The goal of this notebook is to build a model that predicts the `income` level from the features included in the dataset (education level, marital status, occupation, and so on). The first step is to manipulate, or preprocess, the features so they are in the format MLlib requires.  Convert categorical variables to numericSome machine learning algorithms, such as linear and logistic regression, require numeric features. The Adult dataset includes categorical features such as education, occupation, and marital status. The following code block illustrates how to use `StringIndexer` and `OneHotEncoder` to convert categorical variables into a set of numeric variables that only take on values 0 and 1. - `StringIndexer` converts a column of string values to a column of label indexes. For example, it might convert the values "red", "blue", and "green" to 0, 1, and 2. - `OneHotEncoder` maps a column of category indices to a column of binary vectors, with at most one "1" in each row that indicates the category index for that row.One-hot encoding in Spark is a two-step process. You first use the StringIndexer, followed by the OneHotEncoder. The following code block defines the StringIndexer and OneHotEncoder but does not apply it to any data yet.For more information:   [StringIndexer](http://spark.apache.org/docs/latest/ml-features.htmlstringindexer)   [OneHotEncoder](https://spark.apache.org/docs/latest/ml-features.htmlonehotencoder)<jupyter_code>from pyspark.ml.feature import OneHotEncoder, StringIndexer


categoricalCols = [
    "workclass",
    "education",
    "marital_status",
    "occupation",
    "relationship",
    "race",
    "sex",
]

# The following two lines are estimators. They return functions that we will later apply to transform the dataset.
stringIndexer = StringIndexer(
    inputCols=categoricalCols,
    outputCols=[x + "Index" for x in categoricalCols],
)
encoder = OneHotEncoder(
    inputCols=stringIndexer.getOutputCols(),
    outputCols=[x + "OHE" for x in categoricalCols],
)

# The label column ("income") is also a string value - it has two possible values, "<=50K" and ">50K".
# Convert it to a numeric value using StringIndexer.
labelToIndex = StringIndexer(inputCol="income", outputCol="label")<jupyter_output><empty_output><jupyter_text>In this notebook, we'll build a pipeline combining all of our feature engineering and modeling steps. But let's take a minute to look more closely at how estimators and transformers work by applying the `stringIndexer` estimator that we created in the previous code block.You can call the `.fit()` method to return a `StringIndexerModel`, which you can then use to transform the dataset. The `.transform()` method of `StringIndexerModel` returns a new DataFrame with the new columns appended. Scroll right to see the new columns if necessary. For more information: [StringIndexerModel](https://spark.apache.org/docs/latest/api/java/org/apache/spark/ml/feature/StringIndexerModel.html)<jupyter_code>stringIndexerModel = stringIndexer.fit(trainDF)
display(stringIndexerModel.transform(trainDF))<jupyter_output><empty_output><jupyter_text>Combine all feature columns into a single feature vectorMost MLlib algorithms require a single features column as input. Each row in this column contains a vector of data points corresponding to the set of features used for prediction. MLlib provides the `VectorAssembler` transformer to create a single vector column from a list of columns.The following code block illustrates how to use VectorAssembler.For more information: [VectorAssembler](https://spark.apache.org/docs/latest/ml-features.htmlvectorassembler)<jupyter_code>from pyspark.ml.feature import VectorAssembler


# This includes both the numeric columns and the one-hot encoded binary vector columns in our dataset.
numericCols = [
    "age",
    "fnlwgt",
    "education_num",
    "capital_gain",
    "capital_loss",
    "hours_per_week",
]
assemblerInputs = [c + "OHE" for c in categoricalCols] + numericCols
vecAssembler = VectorAssembler(inputCols=assemblerInputs, outputCol="features")<jupyter_output><empty_output><jupyter_text>Step 3. Define the modelThis notebook uses a [logistic regression](https://spark.apache.org/docs/latest/ml-classification-regression.htmllogistic-regression) model.<jupyter_code>from pyspark.ml.classification import LogisticRegression


lr = LogisticRegression(featuresCol="features", labelCol="label", regParam=1.0)<jupyter_output><empty_output><jupyter_text>Step 4. Build the pipelineA `Pipeline` is an ordered list of transformers and estimators. You can define a pipeline to automate and ensure repeatability of the transformations to be applied to a dataset. In this step, we define the pipeline and then apply it to the test dataset.Similar to what we saw with `StringIndexer`, a `Pipeline` is an estimator. The `pipeline.fit()` method returns a `PipelineModel`, which is a transformer.For more information:   [Pipeline](https://spark.apache.org/docs/latest/api/python/pyspark.ml.htmlpyspark.ml.Pipeline)  [PipelineModel](https://spark.apache.org/docs/latest/api/java/org/apache/spark/ml/PipelineModel.html)<jupyter_code>from pyspark.ml import Pipeline


# Define the pipeline based on the stages created in previous steps.
pipeline = Pipeline(stages=[stringIndexer, encoder, labelToIndex, vecAssembler, lr])

# Define the pipeline model.
pipelineModel = pipeline.fit(trainDF)

# Apply the pipeline model to the test dataset.
predDF = pipelineModel.transform(testDF)<jupyter_output><empty_output><jupyter_text>Display the predictions from the model. The `features` column is a [sparse vector](https://spark.apache.org/docs/latest/api/python/pyspark.ml.htmlpyspark.ml.linalg.SparseVector), which is often the case after one-hot encoding, because there are so many 0 values.<jupyter_code>display(predDF.select("features", "label", "prediction", "probability"))<jupyter_output><empty_output><jupyter_text>Step 5. Evaluate the modelThe `display` command has a built-in ROC curve option.<jupyter_code>display(
    pipelineModel.stages[-1],
    predDF.drop("prediction", "rawPrediction", "probability"),
    "ROC",
)<jupyter_output><empty_output><jupyter_text>To evaluate the model, we use the `BinaryClassificationEvaluator` to evalute the area under the ROC curve and the `MulticlassClassificationEvaluator` to evalute the accuracy.For more information:  [BinaryClassificationEvaluator](https://spark.apache.org/docs/latest/api/python/pyspark.ml.htmlpyspark.ml.evaluation.BinaryClassificationEvaluator)  [MulticlassClassificationEvaluator](https://spark.apache.org/docs/latest/api/python/pyspark.ml.htmlpyspark.ml.evaluation.MulticlassClassificationEvaluator)<jupyter_code>from pyspark.ml.evaluation import (
    BinaryClassificationEvaluator,
    MulticlassClassificationEvaluator,
)


bcEvaluator = BinaryClassificationEvaluator(metricName="areaUnderROC")
print(f"Area under ROC curve: {bcEvaluator.evaluate(predDF)}")

mcEvaluator = MulticlassClassificationEvaluator(metricName="accuracy")
print(f"Accuracy: {mcEvaluator.evaluate(predDF)}")<jupyter_output><empty_output><jupyter_text>Step 6. Hyperparameter tuningMLlib provides methods to facilitate hyperparameter tuning and cross validation. - For hyperparameter tuning, `ParamGridBuilder` lets you define a grid search over a set of model hyperparameters.- For cross validation, `CrossValidator` lets you specify an estimator (the pipeline to apply to the input dataset), an evaluator, a grid space of hyperparameters, and the number of folds to use for cross validation.  For more information:   [Model selection using cross-validation](https://spark.apache.org/docs/latest/ml-guide.htmlexample-model-selection-via-cross-validation)   [ParamGridBuilder](https://spark.apache.org/docs/latest/ml-tuning.html)   [CrossValidator](https://spark.apache.org/docs/latest/api/python/pyspark.ml.htmlpyspark.ml.tuning.CrossValidator) Use `ParamGridBuilder` and `CrossValidator` to tune the model. This example uses three values for `regParam` and three for `elasticNetParam`, for a total of 3 x 3 = 9 hyperparameter combinations for `CrossValidator` to examine.<jupyter_code>from pyspark.ml.tuning import CrossValidator, ParamGridBuilder


paramGrid = (
    ParamGridBuilder()
    .addGrid(lr.regParam, [0.01, 0.5, 2.0])
    .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0])
    .build()
)<jupyter_output><empty_output><jupyter_text>Whenever you call `CrossValidator` in MLlib, Databricks automatically tracks all of the runs using [MLflow](https://mlflow.org/). You can use the MLflow UI ([AWS](https://docs.databricks.com/applications/mlflow/index.html)|[Azure](https://docs.microsoft.com/azure/databricks/applications/mlflow/)) to compare how each model performed.In this example we use the pipeline we created as the estimator.<jupyter_code># Create a 3-fold CrossValidator
cv = CrossValidator(
    estimator=pipeline,
    estimatorParamMaps=paramGrid,
    evaluator=bcEvaluator,
    numFolds=3,
    parallelism=4,
)

# Run cross validations. This step takes a few minutes and returns the best model found from the cross validation.
cvModel = cv.fit(trainDF)<jupyter_output><empty_output><jupyter_text>Step 7. Make predictions and evaluate model performanceUse the best model identified by the cross-validation to make predictions on the test dataset, and then evaluate the model's performance using the area under the ROC curve.<jupyter_code># Use the model identified by the cross-validation to make predictions on the test dataset
cvPredDF = cvModel.transform(testDF)

# Evaluate the model's performance based on area under the ROC curve and accuracy
print(f"Area under ROC curve: {bcEvaluator.evaluate(cvPredDF)}")
print(f"Accuracy: {mcEvaluator.evaluate(cvPredDF)}")<jupyter_output><empty_output><jupyter_text>Using SQL commands, you can also display predictions grouped by age and occupation. This requires creating a temporary view of the predictions dataset.<jupyter_code>cvPredDF.createOrReplaceTempView("finalPredictions")
# %sql
# SELECT occupation, prediction, count(*) AS count
# FROM finalPredictions
# GROUP BY occupation, prediction
# ORDER BY occupation
# %sql
# SELECT age, prediction, count(*) AS count
# FROM finalPredictions
# GROUP BY age, prediction
# ORDER BY age<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08795906924586487, 'fraction_numerical': 0.004765909728062798, 'mean_word_length': 6.035996055226825, 'pattern_counts': {'":': 0, '<': 74, '<?xml version=': 0, '>': 74, 'https://': 19, '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': '32707', 'license': 'MIT', 'n_tokens_mistral': 4152, 'n_tokens_neox': 3918, 'n_words': 1401, 'path': 'big-data-school-notebooks/Getting Started with MLlib.ipynb', 'repo_name': 'Code360In/spark-code-examples', 'size': '14268'}