python之在 Spark 和 Python 中使用决策树算法进行分析的问题

我正在对电信行业进行流失分析，并且我有一个示例数据集。我在下面通过 python 在 Spark 中使用 decision tree 算法编写了这段代码。在数据集中，我有多个列，我正在选择我的 feature 集所需的列。

from pyspark.mllib.regression import LabeledPoint 
from pyspark.mllib.tree import DecisionTree, DecisionTreeModel 
from pyspark.mllib.util import MLUtils 
import os.path 
import numpy as np 
 
 
inputPath = os.path.join('file1.csv') 
file_name = os.path.join(inputPath) 
data = sc.textFile(file_name).zipWithIndex().filter(lambda (line,rownum): rownum>0).map(lambda (line, rownum): line) 
 
 
final_data = data.map(lambda line: line.split(",")).filter(lambda line: len(line)>1).map(lambda line:LabeledPoint(1 if line[5] == 'True' else 0,[line[6],line[7]])) 
 
(trainingdata, testdata) = final_data.randomSplit([0.7, 0.3]) 
 
model = DecisionTree.trainRegressor(trainingdata, categoricalFeaturesInfo={}, 
                                    impurity='variance', maxDepth=5, maxBins=32) 
 
predictions = model.predict(testdata.map(lambda x: x.features)) 
prediction= predictions.collect() 
 
labelsAndPredictions = testData.map(lambda lp: lp.label).zip(predictions) 

现在这段代码可以正常工作并进行预测，但我缺少的是 prediction 集合或 testdata 中每个客户的标识符。在我的数据集中有一列用于 customerid(第 4 列)，到目前为止我没有选择它，因为它不是模型中要考虑的功能。我很难将此 customerid 列与详细信息在 testdata 中的客户的 testdata 相关联。如果我在 LabeledPoint 中形成的 feature 向量中添加此从数据集中选择此列，那么这将导致错误，因为它不是特征值。

我如何在我的分析中添加此列，以便我可以获得具有较高流失值(value)的前 50 名客户？

请您参考如下方法:

您可以按照与预测后添加标签完全相同的方式进行操作。

小 helper :

customerIndex = ... # Put index of the column 
 
def extract(line): 
    """Given a line create a tuple (customerId, labeledPoint)""" 
    label = 1 if line[5] == 'True' else 0 
    point =  LabeledPoint(label, [line[6], line[7]]) 
    customerId = line[customerIndex] 
    return (customerId, point) 

使用extract 函数准备日期:

final_data = (data 
    .map(lambda line: line.split(",")) 
    .filter(lambda line: len(line) >1 ) 
    .map(extract)) # Map to tuples 

火车:

# As before 
(trainingdata, testdata) = final_data.randomSplit([0.7, 0.3]) 
 
# Use only points, put the rest of the arguments in place of ... 
model = DecisionTree.trainRegressor(trainingdata.map(lambda x: x[1]), ...) 

预测:

# Make predictions using points 
predictions = model.predict(testdata.map(lambda x: x[1].features)) 
 
# Add customer id and label 
labelsIdsAndPredictions = (testData 
    .map(lambda x: (x[0], x[1].label)) 
    .zip(predictions)) 

提取前 50 名:

top50 = labelsIdsAndPredictions.top(50, key=lambda x: x[1])