创建决策树作为聚合函数而不嵌套 RDD答案

【问题标题】：Create decision tree as aggregate function without nesting RDD创建决策树作为聚合函数而不嵌套 RDD
【发布时间】：2016-02-23 06:18:37
【问题描述】：

我有一大组包含不唯一 ID 和特征向量的行。我需要聚合每个 ID 的所有特征，并使用所有这些特征为每个 ID 生成决策树。

我已经尝试了各种使用 RDD 或数据帧的解决方法，但我总是陷入死胡同。据我所知，不允许嵌套 RDD，所以我无法获得必要的 RDD[LabeledPoint] 来训练地图函数中的树。

具有类似

的结构

idFeatures: RDD[(String, Iterable[LabeledPoint])]

我想做的是与此类似的事情（由于嵌套 RDD 而返回 NullPointerException）：

  val trees = idFeatures.groupByKey().mapValues(features => {
     DecisionTree.trainRegressor(sc.parallelize(features.to[Seq]),  Map[Int, Int](), impurity, maxDepth, maxBins)
  });

有没有一种方法可以做到这一点，而无需遍历整个不同的 id 集并一一生成树（不是考虑到 id 集的大小的选择）？

【问题讨论】：

你有多少数据？每个 id 有多少个不同的 id 和点数？ W
我有大约 4000 万个 id，每个 id 的点数在 1 到 500 之间。考虑到它们并不多，我正在考虑自己实现树构造，但我宁愿使用mlib提供的那些

标签： apache-spark decision-tree rdd apache-spark-mllib

【解决方案1】：

研究使用 RDD。treeAggregate

def treeAggregate[U: ClassTag](zeroValue: U)(
      seqOp: (U, T) => U,
      combOp: (U, U) => U

您可以将trainRegressor放在Sequential Op（第一个参数）中，然后将汇总操作放在Combine Op（第二个参数）中

【讨论】：

我不确定我是否理解这个答案。我想过类似的事情，但嵌套 RDD 仍然存在问题。 trainRegressor 函数需要将完整分区作为每个调用的标记点的 RDD，因此我不能将其用作顺序操作（除非我实现一个 trainRegressor 函数，将每个特征单独添加到树中）。