scala – RDD中是否有任何操作可以保持订单？

发布时间：2020-12-16 18:46:30 所属栏目：安全来源：网络整理

导读：我想在RDD性能中执行一些操作,例如reduce,但不需要操作符是可交换的.即我希望以下结果始终为“123456789”. scala val rdd = sc.parallelize(1 to 9 map (_.toString))rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[24] at parallelize at

我想在RDD性能中执行一些操作,例如reduce,但不需要操作符是可交换的.即我希望以下结果始终为“123456789”.

scala> val rdd = sc.parallelize(1 to 9 map (_.toString))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[24] at parallelize at <console>:24

scala> val result = rdd.someAction{ _+_ }

首先,我找到了折叠. RDD#fold的文档说：

def fold(zeroValue: T)(op: (T,T) ? T): T Aggregate the elements of
each partition,and then the results for all the partitions,using a
given associative function and a neutral “zero value”

请注意,doc中不需要交换.但是,结果并不像预期的那样：

scala> rdd.fold(""){ _+_ }
res10: String = 312456879

编辑我试过@ dk14提到的,没有运气：

scala> val rdd = sc.parallelize(1 to 9 map (_.toString))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[48] at parallelize at <console>:24

scala> rdd.zipWithIndex.map(x => (x._2,x._1)).sortByKey().map(_._2).fold(""){ _+_ }
res22: String = 341276895

scala> rdd.zipWithIndex.map(x => (x._2,x._1)).sortByKey().map(_._2).fold(""){ _+_ }
res23: String = 914856273

scala> rdd.zipWithIndex.map(x => (x._2,x._1)).sortByKey().map(_._2).fold(""){ _+_ }
res24: String = 742539618

scala> rdd.zipWithIndex.map(x => (x._2,x._1)).sortByKey().map(_._2).fold(""){ _+_ }
res25: String = 271468359

解决方法

在Scala中没有满足此条件的内置reduce操作,但您可以通过组合mapPartitions,collect和local reduce来轻松实现自己的操作：

import scala.reflect.ClassTag

def orderedFold[T : ClassTag](rdd: RDD[T])(zero: T)(f: (T,T) => T): T = {
  rdd.mapPartitions(iter => Iterator(iter.foldLeft(zero)(f))).collect.reduce(f)
}

使用collect和reduce组合而不是fold使用的异步和无序方法可确保保留全局顺序.

这当然需要一些额外费用,包括：

>驱动程序的内存占用略高.>显着更高的延迟 – 我们在开始本地缩减之前明确等待所有任务完成.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!