将 spark 的 MLLib 例程与 pandas 数据帧一起使用

Question

我有一个相当大的数据集（~20GB）作为 Pandas/PyTables HDFStore 存储在磁盘上，我想在上面运行随机森林和增强树。尝试在我的本地系统上执行此操作需要很长时间，因此我正在考虑将其移植到我可以访问的 spark 集群，而不是使用 MLLib 例程。

虽然我设法将 pandas 数据帧加载为 spark 数据帧，但我对如何在 MLLib 例程中使用它感到有些困惑。我对 MLLib 不太熟悉，它似乎只接受 LabeledPoint 数据类型。

如果有任何想法/指针/代码解释如何使用（pandas 或 spark）数据帧作为 MLLib 算法的输入 - 直接或间接，通过转换为支持的类型，我将不胜感激。

谢谢。

Answer 1

您需要将DataFrame 转换为RDD[LabeledPoint]。注意LabeledPoint 只是(label: Double, features: Vector)。考虑一个从每一行获取值的映射例程：

val rdd = df.map { row =>
  new LabeledPoint(row(0), DenseVector(row.getDouble(1),..., row.getDouble(n)))
}

这将返回一个RDD[LabeledPoint]，例如，您可以将其输入到RandomForest.trainRegressor(...)。详情请查看DataFrame API。

Answer 2

y_train,X_train 在 pandas 数据帧中，将其转换为 mllib 输入数据格式

1. 转换成numpy数组

y_train=np.array(y_train)
X_train=np.array(X_train)

1. 转换成rdd数据格式

from pyspark.mllib.regression import LabeledPoint
train_data=[]
for i in range(X_train.shape[0]):                                                                   
          train_data.append( LabeledPoint(y_train[i],X_train[i]))

1. 并行化它

train_data_rdd=sparkContext.parallelize(train_data)

{for spark context ->
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('pandasToSparkDF').getOrCreate()

spark = SparkSession \
    .builder \
    .getOrCreate()
sparkContext=spark.sparkContext
}

model = GradientBoostedTrees.trainRegressor(train_data_rdd,categoricalFeaturesInfo={}, numIterations=3)