Revision history [back]

PySpark MLlib does not have a built-in implementation of XGBoost, but you can use the XGBoost4j-Spark library to perform cross-validation and tune hyperparameters for XGBoost. Here are the steps to do so:

1. Install XGBoost4j-Spark library using Maven or download the jar file from the website.
2. Load your data into a PySpark DataFrame.
3. Split the data into training and testing sets using Spark's randomSplit() function.
4. Define the XGBoost estimator with the desired hyperparameters to be tuned.
5. Create a pipeline with the XGBoost estimator and any necessary data preprocessing steps.
6. Define the parameter grid containing the hyperparameters to be tuned and their possible values.
7. Define your evaluation metric using PySpark's MulticlassClassificationEvaluator or RegressionEvaluator.
8. Use the CrossValidator class to perform cross-validation with the parameter grid, pipeline, and evaluation metric.
9. Retrieve the best hyperparameters from the CrossValidator object and fit the final model using those hyperparameters.
10. Evaluate the final model on the testing set.

Here is an example code snippet that shows how to perform cross-validation and hyperparameter tuning for XGBoost using PySpark:

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from ml.dmlc.xgboost4j.scala.spark import XGBoostClassifier

# Load data into a PySpark DataFrame
data = spark.read.format("csv").option("header", "true").load("path/to/data.csv")

# Split data into training and testing sets
train, test = data.randomSplit([0.7, 0.3], seed=123)

# Define the XGBoost estimator with the desired hyperparameters to be tuned
xgb = XGBoostClassifier(objective="binary:logistic")

# Create a pipeline with the XGBoost estimator and any necessary data preprocessing steps
assembler = VectorAssembler(inputCols=["col1", "col2", "col3"], outputCol="features")
pipeline = Pipeline(stages=[assembler, xgb])

# Define the parameter grid containing the hyperparameters to be tuned and their possible values
paramGrid = ParamGridBuilder() \
    .addGrid(xgb.maxDepth, [3, 5, 7]) \
    .addGrid(xgb.eta, [0.1, 0.01, 0.001]) \
    .build()

# Define your evaluation metric using PySpark's BinaryClassificationEvaluator
evaluator = BinaryClassificationEvaluator(labelCol="label", rawPredictionCol="rawPrediction", metricName="areaUnderROC")

# Use the CrossValidator class to perform cross-validation with the parameter grid, pipeline, and evaluation metric
cv = CrossValidator(estimator=pipeline, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=5)
cvModel = cv.fit(train)

# Retrieve the best hyperparameters from the CrossValidator object and fit the final model using those hyperparameters
bestModel = cvModel.bestModel
bestParams = bestModel.stages[-1].extractParamMap()
xgb.setParams(**bestParams)
finalModel = pipeline.fit(train)

# Evaluate the final model on the testing set
predictions = finalModel.transform(test)
auc = evaluator.evaluate(predictions)
print("AUC on testing set: {:.4f}".format(auc))