Пустой массив, возвращаемый mapPartitions в PySpark

Question

привет, может кто-нибудь объяснить мне, почему mapPartitions реагирует по-разному на эти две функции? (Я смотрел на эту this thread, и я не думаю, что моя проблема возникает из-за того, что мой итерацию является TraversableOnce как я его создать.

L=range(10) 
J=range(5,15) 
K=range(8,18) 

data=J+K+L 

def function_1(iter_listoflist): 
    final_iterator=[] 
    for sublist in iter_listoflist: 
     final_iterator.append([x for x in sublist if x%9!=0]) 
    return iter(final_iterator) 

def function_2(iter_listoflist): 
    final_iterator=[] 
    listoflist=list(iter_listoflist) 
    for i in range(len(listoflist)): 
     for j in range(i+1,len(listoflist)): 
      sublist=listoflist[i]+listoflist[j] 
      final_iterator.append([x for x in sublist if x%9!=0]) 
      pass 
     pass 
    return iter(final_iterator) 



sc.parallelize(data,3).glom().mapPartitions(function_1).collect() 

возвращает то, что он должен в то время как

sc.parallelize(data,3).glom().mapPartitions(function_2).collect() 

возвращает пустой массив, я проверяемый код, возвращая список в конце, и он делает то, что я хочу его.

спасибо за вашу помощь

P hilippe C

Answer 1

На самом деле это довольно просто. listoflist всегда имеет длину, равную 1. Чтобы понять, почему это так, вам нужно подумать о том, что происходит, когда звоните glom. Цитирую the docs возвращается:

РДУ, созданный сливающихся все элементы внутри каждого раздела в список.

Это означает, что, когда вы звоните:

listoflist=list(iter_listoflist) 

Вы получаете список со списком одного элемента, содержащего все элементы из этого раздела. Игнорирование все детали:

(sc.parallelize(data, 3) 
    .glom() 
    .mapPartitionsWithIndex(lambda i, iter: [(i, list(iter))]) 
    .collect()) 

## [(0, [[5, 6, 7, 8, 9, 10, 11, 12, 13, 14]]), 
##  (1, [[8, 9, 10, 11, 12, 13, 14, 15, 16, 17]]), 
##  (2, [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]])] 

Это означает, что

• range(len(listoflist)) содержит единственный элемент, равный 0.
• range(i+1,len(listoflist)) путем замещения пустой range(1, 1)

Поэтому нет ничего do и вы получите пустой итератор.

С одной стороны все эти сообщения pass, а также звонки iter являются полностью устаревшими.