这里使用了Pipeline来封装特征处理和模型训练步骤#xff0c;保存为…遇到一个还不错的使用Xgboost训练模型的githubhttps://github.com/MachineLP/Spark-/tree/master/pyspark-xgboost

这里使用了Pipeline来封装特征处理和模型训练步骤保存为pipelineModel。

pipeline

SparkSessionos.environ[PYSPARK_PYTHON]

Python3.7/bin/python

os.environ[PYSPARK_SUBMIT_ARGS]

--jars

xgboost4j-spark-0.90.jar,xgboost4j-0.90.jar

pyspark-shellspark

\.config(spark.driver.allowMultipleContexts,

true)

\.config(spark.yarn.dist.archives,

hdfs://ns62007/user/dmc_adm/_PYSPARK_ENV/Python3.7.zip#Python3.7)

\.config(spark.executorEnv.PYSPARK_PYTHON,

\.config(spark.sql.autoBroadcastJoinThreshold,

-1)

\.getOrCreate()spark.sparkContext.addPyFile(sparkxgb.zip)schema

StructType([StructField(PassengerId,

DoubleType()),StructField(Survived,

DoubleType()),StructField(Pclass,

DoubleType()),StructField(Name,

DoubleType()),StructField(SibSp,

DoubleType()),StructField(Parch,

DoubleType()),StructField(Ticket,

StringType()),StructField(Fare,

DoubleType()),StructField(Cabin,

StringType()),StructField(Embarked,

titanic/train.csv

hdfs://tmp/gao/dev_data/dmb_upload_data/

file_path

upload_file.split(/)[-1])df_raw

true)\.schema(schema)\.csv(file_path)df_raw.show(20)

StringIndexer()\.setInputCol(Sex)\.setOutputCol(SexIndex)\.setHandleInvalid(keep)cabinIndexer

StringIndexer()\.setInputCol(Cabin)\.setOutputCol(CabinIndex)\.setHandleInvalid(keep)embarkedIndexer

StringIndexer()\.setInputCol(Embarked)\.setHandleInvalid(keep)#

.setOutputCol(EmbarkedIndex)\vectorAssembler

VectorAssembler()\.setInputCols([Pclass,

Age,

Fare])\.setOutputCol(features)from

sparkxgb

XGBoostClassifier(maxDepth3,missingfloat(0.0),featuresColfeatures,labelColSurvived

)pipeline

Pipeline(stages[vectorAssembler,

xgboost])trainDF,

model.transform(testDF).select(col(PassengerId),

col(Survived),

model.write().overwrite().save(os.path.join(hdfs_path,xgboost_class_test))from

pyspark.ml

PipelineModel.load(os.path.join(hdfs_path,xgboost_class_test))

model1.transform(testDF).show()这是执行结果

2、当然也可以不用pipeline封装直接训练xgboost模型并保存。

Train

XGBoostClassifier(maxDepth3,missingfloat(0.0),featuresColfeatures,

labelColSurvived)#

trained_raw_model.transform(td)

rawPrediction,

trained_raw_model.nativeBooster.saveModel(outputmodel.xgboost)#

from

XGBoostClassifier,XGBoostClassificationModel

model1

XGBoostClassificationModel.load(outputmodel.xgboost)

这里报错无法使用

XGBoostClassificationModel加载已经训练好的XGB模型。