pyspark mapPartitions 함수는 어떻게 작동합니까?

Question

그래서 파이크 (Pyspark)를 사용하여 스파크를 배우려고합니다. 나는 기능이 mapPartitions 어떻게 작동하는지 알고 싶다. 그것이 Input이 취하는 것과 Output이주는 것입니다. 인터넷에서 적절한 예를 찾을 수 없었습니다. 말하자면, 아래 목록과 같은 목록을 포함하는 RDD 객체가 있습니다.

[ [1, 2, 3], [3, 2, 4], [5, 2, 7] ] 

그리고 나는 모든 목록에서 요소 (2)를 제거하려면, 어떻게 내가 mapPartitions를 사용하는 것을 달성 할 것입니다.

Answer 1

mapPartition은 파티션의 맵 작업이 아닌 파티션의 맵 조작으로 간주되어야합니다. 그것은 입력의 현재 파티션의 집합입니다 출력은 다른 파티션의 집합이 될 것입니다.

당신이지도는 mapPartition이 RDD 유형의 반복자를 수행해야 통과하여 RDD

기능의 개별 요소를 가지고 다른 또는 같은 종류의 반환과 반복 가능해야한다 전달하는 기능. 귀하의 경우에는

당신은 아마 그냥 그것은 yield를 사용하여 발전기 기능을 mapPartitions를 사용하는 것이 더 쉽습니다

def filterOut2FromPartion(list_of_lists): 
    final_iterator = [] 
    for sub_list in list_of_lists: 
    final_iterator.append([x for x in sub_list if x != 2]) 
    return iter(final_iterator) 

filtered_lists = data.mapPartition(filterOut2FromPartion) 

Answer 2

될 것이다 당신이 mapPartition을 사용하기를 원한다면

def filterOut2(line): 
    return [x for x in line if x != 2] 

filtered_lists = data.map(filterOut2) 

같은 것을하고 싶어 구문 :

def filter_out_2(partition): 
    for element in partition: 
     if element != 2: 
      yield element 

filtered_lists = data.mapPartition(filter_out_2)