如何将 Scalding ValuePipe 加入 TypedPipe？答案

【问题标题】：how can I join a Scalding ValuePipe to a TypedPipe?如何将 Scalding ValuePipe 加入 TypedPipe？
【发布时间】：2015-12-18 16:15:28
【问题描述】：

我已经修改了 scalding KMeans 示例来执行 KModes。问题是当作业完成时，我需要将聚类记录与匹配的质心连接起来。 KMeans 代码使用 ValuePipe 来保存质心。因此，为了将质心从 ValuePipe 中取出，我正在对其进行平面映射。然后我像这样加入：

  HVKModes(500000,inputSets,10).waitFor(Config.default,mode) match {
    case Success((a,centroids: ValuePipe[List[LabeledCentroid]], points: TypedPipe[LabeledVector])) => {
      val joined = centroids
        .flatMap {
          cs : List[LabeledCentroid] => {
          val t = TypedPipe.from(cs)
          Iterator(points.join(t)) }
        }
        .values
        .write(clusteredOutput)
    }
    case Failure(e) => sys.error("problem running job:" + e.toString)
  }

问题是编译器在“值”行给出了类型错误：

Cannot prove that com.twitter.scalding.typed.CoGrouped[Int,((String, Set[String]), Set[String])] <:< (Any, V).
[error]         .values

我收集到错误表明它无法计算出我用来表示值的 V。但是我该怎么办呢？

【问题讨论】：

标签： scala hadoop cluster-analysis scalding

【解决方案1】：

我几乎是对的。我只需要加入平面映射的结果。

  HVKModes(500000,inputSets,10).waitFor(Config.default,mode) match {
    case Success((a,centroids: ValuePipe[List[LabeledCentroid]], points: TypedPipe[LabeledVector])) => {

      val cs  =
        centroids
        .flatMap {
          cs : List[LabeledCentroid] => { cs.toIterator}
        }

      points
        .join(cs)
        .values
        .write(clusteredOutput)
    }
    case Failure(e) => sys.error("problem running job:" + e.toString)
  }

【讨论】：