Spark RDD转换问题

Question

我有这种格式的数据：

100 1 2 3 4 5

我使用以下代码来加载它：

 val data : RDD[(String, Array[Int])] = sc.textFile("data.txt").map(line => ((line.split("\t"))(0), (line.split("\t"))(1).split(" ").map(_.toInt)))

我想从数组[Int]生成对，这样一个值大于一个数（在下面的代码中为2）的数组元素就可以与数组的所有其他元素配对。然后我会用它来生成更多的统计数据。例如，对于示例数据，我应该能够首先生成：

100 (3,1), (3,2), (3,4), (3,5),(4,1), (4,2), (4,3), (4,5)

val test = merged_data.mapValues { case x =>
      for (element <- x) {
        val y = x.filter(_ != element)

        if (element > 2)
          {

            for (yelement <- y)
              {
                (element, yelement)
              }
          }
      }
      }

下面是我得到的o/p：Array[（String，Unit）]=Array（（100，（））不确定为什么它是空的。

一旦我能够解决这个问题，我将对元组中的元素进行排序，并删除重复的（如果有的话）

100 (3,1), (3,2), (3,4), (3,5),(4,1), (4,2), (4,3), (4,5)

变成这样：

100 (1,3), (2,3), (3,4), (3,5), (1,4), (2,4), (4,5)

Answer 1

比如：

val test = data.mapValues { x =>
    for {
        element <- x.filter(_ > 2);
        yelement <- x.filter(_ != element)
    } yield (element, yelement)
}

您可能还想查看：Scala中的嵌套迭代，它回答了为什么会得到一个空结果。

Answer 2

我能够解决这个问题，因为：

  val test = merged_data.mapValues { case x =>
  var sb = new StringBuilder

  for (element <- x) {
    val y = x.filter(_ != element)

    if (element > 2)
      {

        for (yelement <- y)
          {
            (element, yelement)
          }
      }
  }
  sb.toString()
  }