Wykorzystanie pamięci Spark/przędzy Python

Question

Mam aplikację świecy zapłonowej, która jest zabijana przez przędzę za przekroczenie limitów pamięci. Mam krok, który wymaga załadowania niektórych zasobów, które są nieco ciężkie (500+ MB), więc używam mapPartitions. Zasadniczo:

def process_and_output(partition): 
    resources = load_resources() 
    for record in partition: 
     yield transform_record(resources, record) 

input = sc.textFile(input_location) 
processed = input.mapPartitions(process_and_output) 
processed.saveAsTextFile(output_location) 

Kiedy pracuje, ja konsekwentnie dostać ten błąd:

ERROR YarnScheduler: Przegrane wykonawca 1 na (adres usunięte): Pojemnik zabity przez przędzy za przekroczenie limitów pamięci. Wykorzystano 11,4 GB pamięci fizycznej 11,2 GB. Rozważ wzmocnienie spark.yarn.executor.memoryOverhead.

Próbowałem podwyższać wartość parametru MemoryOver do bardzo wysokiego, ale wciąż tego samego problemu. Pobiegłem z:

--conf "spark.python.worker.memory=1200m" \ 
--conf "spark.yarn.executor.memoryOverhead=5300" \ 
--conf "spark.executor.memory=6g" \ 

Z pewnością wystarczy pamięci MemoryOverhead?

Chyba bardziej ogólnie, staram się zrozumieć, w jaki sposób pamięć robotów Pythona jest kontrolowana/liczona w ogólnej sumie. Czy jest jakaś dokumentacja tego?

Chciałbym również zrozumieć, czy użycie funkcji generatora zmniejszy zużycie pamięci. Czy będzie przesyłać strumieniowo dane w procesie Pythona (jak mam nadzieję), czy też zbuforuje to wszystko przed wysłaniem z powrotem do infrastruktury JVM/iskry?

Answer 1

Jak widać 11,2 GB to maksymalna pamięć dla kontenera utworzonego przez przędzę. Jest równa pamięci executora + narzut. Tak więc pamięć Pythona nie jest brana pod uwagę.

Wyjątek wymaga zwiększenia narzutu, ale można po prostu zwiększyć pamięć executorów bez zwiększania pamięci głownej. To wszystko, co mogę powiedzieć, nie wiedząc, dlaczego potrzebujesz tyle pamięci w jednym executorze, może być kartezjańskim lub czymś podobnym, może wymagać tak dużo pamięci.

Answer 2

Yarn zabija wykonawców podczas jego

memory usage > (executor-memory + executor.memoryOverhead)

Od Ustawienie to wygląda, że ​​jest to ważny wyjątek.

(memory usage)11.4GB > 11.18GB (executor-memory=6GB + memoryOverhead=5.18GB)

spróbuj

--conf "spark.yarn.executor.memoryOverhead=6144"