Spark - 无重复的组合

Question

我正在尝试在不重复文本文件的情况下进行所有行组合。

例子：

1. 1
2. 2
3. 2
4. 1
5. 1

结果：

• 第 1 行，第 2 行 = (1,2)
• 第 1 行，第 3 行 = (1,2)
• 第 1 行，第 4 行 = (1,1)
• 第 1 行，第 5 行 = (1,1)
• 第 2 行，第 3 行 = (2,2)
• 第 2 行，第 4 行 = (2,1)
• 第 2 行，第 5 行 = (2,1)
• 第 3 行，第 4 行 = (2,1)
• 第 3 行，第 5 行 = (2,1)
• 第 4 行，第 5 行 = (1,1)

或

考虑 (x,y)，如果 (x != y) 0 else 1：

• 0
• 0
• 1
• 1
• 1
• 0
• 0
• 0
• 0
• 1

我有以下代码：

def processCombinations(rdd: RDD[String]) = {
    rdd.mapPartitions({ partition => {
        var previous: String = null;
        if (partition.hasNext)
          previous = partition.next

        for (element <- partition) yield {
          if (previous == element)
            "1"
          else
            "0"
        }
      }
    })
  }

上面的代码是对我的RDD的第一个元素进行组合，换句话说：(1,2) (1,2) (1,1) (1,1)。

问题是：此代码仅适用于一个分区。我想让这个工作在许多分区上，我该怎么做？

Answer 1

您想要输出的确切内容不是很清楚，但这复制了您的第一个示例，并直接转换为 Spark。它生成组合，但仅在原始列表中第一个元素的索引小于第二个元素的索引时，我认为这是您所要求的。

val r = List(1,2,2,1,1)
val z = r zipWithIndex

z.flatMap(x=>z.map(y=>(x,y))).collect{case(x,y) if x._2 < y._2 => (x._1, y._1)}
//List((1,2), (1,2), (1,1), (1,1), (2,2), (2,1), (2,1), (2,1), (2,1), (1,1))

或者，作为一种理解

for (x<-z; y<-z; if x._2 < y._2) yield (x._1, y._1)

Answer 2

此代码使用递归计算不重复的组合。它有 2 个参数：组合的元素数量和元素列表。

它的工作方式如下：对于给定的列表：1、2、3、4、5 =>第一个组合需要 4 个第一个元素。然后它与列表的最后一个元素 5 生成其他组合。当列表中没有更多元素时，它会向后移动一个位置（第三个位置）并获取下一个元素以从那里生成更多组合：1、2、“4”、5。此操作是递归完成的，所有列表的元素。

def combinator[A](n: Int, list: List[A], acc: List[A]): List[List[A]] = {
  if (n == 0)
    List(acc.reverse)
  else if (list == Nil)
    List()
  else
    combinator(n - 1, list.tail, list.head :: acc) ::: combinator(n, list.tail, acc)
}

combinator(4, List(1, 2, 3, 4, 5), List()).foreach(println)

// List(1, 2, 3, 4)
// List(1, 2, 3, 5)
// List(1, 2, 4, 5)
// List(1, 3, 4, 5)
// List(2, 3, 4, 5)