如何在RDD(Scala)上创建嵌套的forloop

Question

我有一个具有以下结构的RDD:
((ByteArray, Idx), ((srcIdx,srcAdress), (destIdx,destAddress)))

这是比特币区块链的边缘(交易)的表示.(ByteArray, Idx)可以看作是一个标识符,休息是一个边缘.我的最终目标是在区块链的图形表示中聚合节点.对此我需要做的结构的第一个修改是将位于同一比特币事务中的源放在一个边缘(最终在一个节点中).通过这种方式,我将"集群"属于同一用户的公钥.此修改的结果将具有以下结构:
((ByteArray, Idx), (List((srcIdx, srcAddress)), (destIdx, destAddress)))
或者以任何其他形式具有相同的功能(例如,如果这在Scala中不可能或不合逻辑).

我目前的思考过程如下.在Java中,我将对RDD中的项进行嵌套for循环,每个循环为具有相同键((ByteArray, Idx))的项创建列表.删除任何重复后.但是,由于我正在处理RDD和Scala,这是不可能的.接下来,我尝试在我的RDD上执行a .collect()然后单独的.map()功能,使用集合在我的map函数中循环.但是,Spark并不喜欢这样,因为显然集合无法序列化.接下来,我尝试创建一个"嵌套"映射函数,如下所示:

val aggregatedTransactions = joinedTransactions.map( f => {
  var list = List[Any](f._2._1)

  val filtered = joinedTransactions.filter(t => f._1 == t._1)

  for(i <- filtered){
    list ::= i._2._1
  }

  (f._1, list, f._2._2)
})

这是不允许的,因为.map()中没有过滤器(或映射)功能.有哪些替代方案？

我对Scala很新,因此非常感谢任何有用的背景信息.

Answer 1

我的最终目标是在区块链的图形表示中聚合节点.对此我需要做的结构的第一个修改是将位于同一比特币事务中的源放在一个边缘(最终在一个节点中).

所以基本上你想要groupByKey:

joinedTransactions.groupByKey().map {
   // process data to get desired shape
}