RDD에 조치가 있습니까?

Question

RDD과 같은 성능을 원하지만 reduce과 같은 성능을 원하지만 연산자를 교환 할 필요는 없습니다. 즉, 나는 result을 다음과 같이 항상 "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{ _+_ } 

우선, fold을 찾았습니다. RDD#fold의 문서를 말한다

DEF (zeroValue : T) 배하여 모든 파티션 요소마다 분할하고, 그 결과를 T 골재 (OP (T, T)의 T를 ⇒) 주어진 연관 함수 중성 "영점"다큐먼트에 필요하지 가환 없다고

참고. 그러나, 결과는 예상과 :

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 

Answer 1

있다, 이것은 원래 RDD 일에만 파티션을 병합 고려 설명하기 위해 어떤 스칼라에서이 기준을 만족 활동을 감소 내장,하지만 당신은 쉽게 mapPartitions, collect 및 지방 감소를 결합하여 자신을 구현할 수 있습니다 :

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) 
} 

fold 의해 사용 병합 대신 비동기 및 정렬되지 않은 방법에 대해 collectreduce 및 조합을 사용하여 글로벌 순서가 보존되는 것을 보장한다. 물론

이 포함 몇 가지 추가 비용으로 제공 :

• 약간 더 높은 메모리 풋 프린트 드라이버에.
• 지연 시간이 훨씬 더 길다. 로컬 감소를 시작하기 전에 모든 작업이 완료 될 때까지 기다린다.

Answer 2

으로는 순서를 보존하지 않습니다 @YuvalItzchakov fold 지적 :

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

EDIT @의 dk14에서 언급 한 바와 같이 나는 운없이, 시도 결과를 조합 할 때 파티션이 RDD 인 경우

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

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

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

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