Spark Scala：如何处理 rdd 的每 3 个元素？

Question

每个人。

我有这样的问题：

我有非常大的 rdd：数十亿个元素，例如：

Array[((Int, Int), Double)] = Array(((0,0),729.0), ((0,1),169.0), ((0,2),1.0), ((0,3),5.0), ...... ((34,45),34.0), .....)

我需要做这样的操作：

通过键(i,j)获取每个元素的值并添加到它

min(rdd_value[(i-1, j)],rdd_value[(i, j-1)], rdd_value[(i-1, j-1)])

如果不使用 collect() 作为 collect() 之后，我怎么能做到这一点，因为我的 rdd 非常大，所以我得到了 Java memory errror。

非常感谢！

我尝试从 python 实现这个算法。当时间序列是 rdds 时。

def DTWDistance(s1, s2):
    DTW={}

    for i in range(len(s1)):
        DTW[(i, -1)] = float('inf')
    for i in range(len(s2)):
        DTW[(-1, i)] = float('inf')
    DTW[(-1, -1)] = 0

    for i in range(len(s1)):
        for j in range(len(s2)):
            dist= (s1[i]-s2[j])**2
            DTW[(i, j)] = dist + min(DTW[(i-1, j)],DTW[(i, j-1)], DTW[(i-1, j-1)])

    return sqrt(DTW[len(s1)-1, len(s2)-1])

现在我应该使用 for 循环执行最后一个操作。 dist 已经计算好了。

示例：

输入（如矩阵）：

4 5 1
7 2 3
9 0 1

Rdd 看起来像

rdd.take(10)

Array(((1,1), 4), ((1,2), 5), ((1,3), 1), ((2,1), 7), ((2,2), 2), ((2,3), 3), ((3,1), 9), ((3,2), 0), ((3,3), 1))

我想做这个操作

rdd_value[(i, j)] = rdd_value[(i, j)] + min(rdd_value[(i-1, j)],rdd_value[(i, j-1)], rdd_value[(i-1, j-1)])

例如：

((1, 1), 4) = 4 + min(infinity, infinity, 0) = 4 + 0 = 4


4 5 1
7 2 3
9 0 1

然后

((1, 2), 5) = 5 + min(infinity, 4, infinity) = 5 + 4 = 9


4 9 1
7 2 3
9 0 1

然后

....

然后

((2, 2), 2) = 2 + min(7, 9, 4) = 2 + 4 = 6


4 9 1
7 6 3
9 0 1

然后 .....

((3, 3), 1) = 1 + min(3, 0, 2) = 1 + 0 = 1

Answer 1

简短的回答是，您尝试解决的问题无法使用 Spark 高效简洁地表达。如果您选择普通 RDD 是分布式矩阵，这并不重要。

要了解为什么您必须考虑 Spark 编程模型。一个基本的 Spark 概念是一个依赖关系图，其中每个 RDD 依赖于一个或多个父 RDD。如果您的问题定义如下：

• 给定一个初始矩阵M⁰
• 对于 i
  • 找到矩阵Mⁱ其中Mⁱ_(m,n) = M^{i - 1}_(m,n) + min(M^{i - 1}_{(m-1,n )}, M^{i - 1}_(m-1,n-1), M^{i - 1}_{(m ,n-1)})

那么使用 Spark API（伪代码）表达将是微不足道的：

rdd
    .flatMap(lambda ((i, j), v): 
        [((i + 1, j), v), ((i, j + 1), v), ((i + 1, j + 1), v)])
    .reduceByKey(min)
    .union(rdd)
    .reduceByKey(add)

不幸的是，您正试图表达同一数据结构中各个值之间的依赖关系。 Spark 抛开它是一个更难并行化的问题，更不用说分发了。

这种类型的动态编程很难并行化，因为在不同的点是完全或几乎完全顺序的。例如，当您尝试计算 Mⁱ_(0,0) 或 Mⁱ_(m,n) 没有什么可以并行化的。它很难分发，因为它可以在块之间产生复杂的依赖关系。

在 Spark 中，通过计算单个块并表达这些块之间的依赖关系或使用迭代算法并在显式图 (GraphX) 上传播消息，有一些重要的方法来处理此问题，但这远非易事。

说到底，对于这种类型的计算而言，相比 Spark，这些工具可能是更好的选择。