RDD를 분할하는 방법

Question

공백으로 구분 된 많은 수의 부동 소수점 값으로 구성된 텍스트 파일이 있습니다. 이 파일을 scala의 RDD에로드하고 있습니다. 이 RDD는 어떻게 파티셔닝됩니까?

또한 모든 파티션의 요소 수와 각 파티션의 인덱스가 같은 사용자 지정 파티션을 생성하는 방법이 있습니까?

val dRDD = sc.textFile("hdfs://master:54310/Data/input*") 
keyval=dRDD.map(x =>process(x.trim().split(' ').map(_.toDouble),query_norm,m,r)) 

여기 HDFS에서 여러 텍스트 파일을로드하고 있습니다. 프로세스는 호출 할 함수입니다. mapPartitonsWithIndex와 함께 솔루션을 사용할 수 있습니까? 어떻게 프로세스 함수 내에서 해당 인덱스에 액세스 할 수 있습니까? 지도는 파티션을 셔플합니다.

Answer 1

당신은 병합 기능을 사용하여 사용자 정의 파티션을 생성 할 수 있습니다

coalesce(numPartitions: Int, shuffle: Boolean = false): RDD[T] 

Answer 2

로드 된 RDD는 기본 파티션 프로그램으로 분할 : 해시 코드를. 사용자 정의 분할기를 지정하려면 자체 분할기와 함께 제공되는 rdd.partitionBy()를 확인할 수 있습니다.

api docs와 같이 coalesce()를 사용하는 것이 좋다고 생각하지 않습니다. coalesce()는 파티션 수를 줄이는 경우에만 사용할 수 있으며 coalesce).

Answer 3

RDD는 어떻게 분할됩니까?

기본적으로 파티션은 각 HDFS 파티션에 대해 만들어지며 기본적으로 64MB입니다. 더 읽기 here.

파티션간에 데이터의 균형을 유지하는 방법은 무엇입니까?

먼저, 하나 다시 분할 자신의 데이터를 할 수있는 세 가지 방법에 대해 살펴 :

1) 두 번째 매개 변수를 전달, (가) textFile()으로 RDD에 대한 파티션 의 최소 번호를 원하는, 하지만 조심 :

In [14]: lines = sc.textFile("data") 

In [15]: lines.getNumPartitions() 
Out[15]: 1000 

In [16]: lines = sc.textFile("data", 500) 

In [17]: lines.getNumPartitions() 
Out[17]: 1434 

In [18]: lines = sc.textFile("data", 5000) 

In [19]: lines.getNumPartitions() 
Out[19]: 5926 

당신이 볼 수 있듯이, [16]는 RDD는이 파티션 수 있기 때문에, 사람이 무엇을 기대하지 않는다, 이미 요청한 최소 파티션 수보다 큽니다.

2)이 같은 repartition(), 사용

In [22]: lines = lines.repartition(10) 

In [23]: lines.getNumPartitions() 
Out[23]: 10 

경고 :이 셔플를 호출하고 증가 당신 RDD는이 파티션의 수를 할 때 사용되어야한다. docs 가입일

:

셔플은 파티션에있어 상이하게 그룹화되도록 재분배 데이터의 점화기구이다. 이것은 전형적으로 집행자와 기계간에 데이터를 복사하여 셔플을 복잡하고 값 비싼 작업으로 만듭니다.

3)이 같은 coalesce()를 사용하여 여기에

In [25]: lines = lines.coalesce(2) 

In [26]: lines.getNumPartitions() 
Out[26]: 2 

는 스파크는 RDD을 축소 할 것이라는 점을 알고 그것의 이점을 얻을 수 있습니다. repartition() vs coalesce()에 대해 더 자세히 읽어보십시오. 데이터가 완벽하게 파티션에서이 모든 보증

---

그러나 균형됩니다 것? 별로 경험하지 못했지만, 내가 경험 한대로 How to balance my data across the partitions?