<jupyter_start><jupyter_text>Introduction to XGBoost Spark with GPUAgaricus is an example of xgboost classifier for multiple classification. This notebook will show you how to load data, train the xgboost model.A few libraries required for this notebook:  1. NumPy  2. cudf jar  3. xgboost4j jar  4. xgboost4j-spark jar  5. rapids-4-spark.jar  This notebook also illustrates the ease of porting a sample CPU based Spark xgboost4j code into GPU. There is only one change required for running Spark XGBoost on GPU. That is replacing the API `setFeaturesCol(feature)` on CPU with the new API `setFeaturesCols(features)`. This also eliminates the need for vectorization (assembling multiple feature columns in to one column) since we can read multiple columns.  Import All Libraries<jupyter_code>from ml.dmlc.xgboost4j.scala.spark import XGBoostClassificationModel, XGBoostClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql import SparkSession
from pyspark.sql.types import FloatType, StructField, StructType
from time import time
import os<jupyter_output><empty_output><jupyter_text>Besides CPU version requires two extra libraries.```Pythonfrom pyspark.ml.feature import VectorAssemblerfrom pyspark.sql.functions import col```  Create Spark Session and Data Reader<jupyter_code>spark = SparkSession.builder.getOrCreate()
reader = spark.read<jupyter_output><empty_output><jupyter_text>Specify the Data Schema and Load the Data<jupyter_code>label = 'label'
features = [ 'feature_' + str(i) for i in range(0, 126) ]
schema = StructType([ StructField(x, FloatType()) for x in [label] + features ])

# You need to update them to your real paths!
dataRoot = os.getenv("DATA_ROOT", "/data")
train_data = reader.schema(schema).option('header', True).csv(dataRoot + '/agaricus/csv/train')
trans_data = reader.schema(schema).option('header', True).csv(dataRoot + '/agaricus/csv/test')<jupyter_output><empty_output><jupyter_text>Note on CPU version, vectorization is required before fitting data to classifier, which means you need to assemble all feature columns into one column.```Pythondef vectorize(data_frame):    to_floats = [ col(x.name).cast(FloatType()) for x in data_frame.schema ]    return (VectorAssembler()        .setInputCols(features)        .setOutputCol('features')        .transform(data_frame.select(to_floats))        .select(col('features'), col(label)))train_data = vectorize(train_data)trans_data = vectorize(trans_data)```  Create a XGBoostClassifier<jupyter_code>params = { 
    'eta': 0.1,
    'missing': 0.0,
    'treeMethod': 'gpu_hist',
    'maxDepth': 2,
    'numWorkers': 1,
    'numRound' : 100,
}
classifier = XGBoostClassifier(**params).setLabelCol(label).setFeaturesCols(features)<jupyter_output><empty_output><jupyter_text>The CPU version classifier provides the API `setFeaturesCol` which only accepts a single column name, so vectorization for multiple feature columns is required.```Pythonclassifier = XGBoostClassifier(**params).setLabelCol(label).setFeaturesCol('features')```The parameter `num_workers` should be set to the number of GPUs in Spark cluster for GPU version, while for CPU version it is usually equal to the number of the CPU cores.Concerning the tree method, GPU version only supports `gpu_hist` currently, while `hist` is designed and used here for CPU training.  Train the Data with Benchmark<jupyter_code>def with_benchmark(phrase, action):
    start = time()
    result = action()
    end = time()
    print('{} takes {} seconds'.format(phrase, round(end - start, 2)))
    return result
model = with_benchmark('Training', lambda: classifier.fit(train_data))<jupyter_output>Training takes 27.95 seconds<jupyter_text>Save and Reload the Model<jupyter_code>model.write().overwrite().save(dataRoot + '/new-model-path')
loaded_model = XGBoostClassificationModel().load(dataRoot + '/new-model-path')<jupyter_output><empty_output><jupyter_text>Transformation and Show Result Sample<jupyter_code>def transform():
    result = loaded_model.transform(trans_data).cache()
    result.foreachPartition(lambda _: None)
    return result
result = with_benchmark('Transformation', transform)
result.select(label, 'rawPrediction', 'probability', 'prediction').show(5)<jupyter_output>Transformation takes 2.63 seconds
+-----+--------------------+--------------------+----------+
|label|       rawPrediction|         probability|prediction|
+-----+--------------------+--------------------+----------+
|  1.0|[-0.9667757749557...|[0.03322422504425...|       1.0|
|  0.0|[-0.0080436170101...|[0.99195638298988...|       0.0|
|  0.0|[-0.0080436170101...|[0.99195638298988...|       0.0|
|  0.0|[-0.1416745483875...|[0.85832545161247...|       0.0|
|  0.0|[-0.0747678577899...|[0.92523214221000...|       0.0|
+-----+--------------------+--------------------+----------+
only showing top 5 rows<jupyter_text>Evaluation<jupyter_code>accuracy = with_benchmark(
    'Evaluation',
    lambda: MulticlassClassificationEvaluator().setLabelCol(label).evaluate(result))
print('Accuracy is ' + str(accuracy))<jupyter_output>Evaluation takes 0.29 seconds
Accuracy is 0.9987577063864658<jupyter_text>Stop<jupyter_code>spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14189965530448104, 'fraction_numerical': 0.04174645729605515, 'mean_word_length': 4.568230277185501, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 34, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '385725', 'license': 'Apache-2.0', 'n_tokens_mistral': 1750, 'n_tokens_neox': 1589, 'n_words': 420, 'path': 'examples/Spark-ETL+XGBoost/agaricus/notebooks/python/agaricus-gpu.ipynb', 'repo_name': 'viadea/spark-rapids-examples', 'size': '5222'}