簡述

　　在用pyspark進行開發的時候，經常會遇到如何將pyspark讀取的資料使用xgboost的模型進行訓練，當然，如果是使用scala，可以直接使用xgboost4j，這個庫裡面提供了可以讀取rdd的資料檔案的函式介面，通過scala可以很簡單的進行訓練，但是對於python使用者來說，如何使用pyspark和xgboost進行訓練，就比較難以處理了？ 　　本人通過對libsvm的資料進行測試，實現了一套通過pyspark讀取資料，xgboost訓練的流程．

　　主要的思路是，首先，讀取libsvm的資料．然後，將pipelineRDD資料轉化為pyspark的DataFrame的資料型別，最後，通過pyspark中DataFrame的資料型別提供的函式，將pyspark中的DataFrame轉化為pandas中的DataFrame的資料型別．

程式碼介紹

讀取libsvm資料型別

data = MLUtils.loadLibSVMFile(sc,'{}://{}'.format(FLAGS.file_path, FLAGS.source_path))
# split data to train , test
(trainingData, testData) = data.randomSplit([0.7, 0.3])

轉化label

def label_rdd_to_dataframe(sqlContext, data):
    label = data.map(lambda x: x.label)
    label_row = Row('label'
 
)
    label_rdd = label.map(lambda r: label_row(r))
    label_DF = sqlContext.createDataFrame(label_rdd)
    return label_DF.toPandas()

轉化features

def features_rdd_to_dataframe(sqlContext, data):
    features = data.map(lambda x: Vectors.dense(x.features.toArray()))
    attribute_str = ['attributes[{}]'
 
.format(i) for i in range(len(features.take(1)[0]))]
    features_rdd = features.map(lambda attributes: [float(eval(value)) for value in attribute_str])
    features_DF = sqlContext.createDataFrame(features_rdd)
    return features_DF.toPandas()

訓練xgboost

# transform train data
label_train_DF = label_rdd_to_dataframe(sqlContext, trainingData)
features_train_DF = features_rdd_to_dataframe(sqlContext,trainingData)
train_data = xgb.DMatrix(features_train_DF, label=label_train_DF)

# transform test data
label_test_DF = label_rdd_to_dataframe(sqlContext, testData)
features_test_DF = features_rdd_to_dataframe(sqlContext, testData)
test_data = xgb.DMatrix(features_test_DF, label= label_test_DF)

# params
param_dict = {'eta':FLAGS.eta,
              "max_depth":FLAGS.max_depth,
              "silent":FLAGS.silent,
              "objective":FLAGS.objective,
              "lambda":FLAGS.xgb_lambda,
              "nthread":FLAGS.nthread}
evallist = [(test_data, 'eval'), (train_data, 'train')]
num_round = FLAGS.num_round

# train
model = xgb.train(param_dict,
                  train_data,
                  num_round,
                  evallist)

完整程式碼連結：