редактировать: Ответ помогает, но я описал свое решение в:ошибка памяти в Spark.

У меня есть RDD с разделами 202092, который читает набор данных, созданный другими. Я могу вручную увидеть, что данные не сбалансированы между разделами, например, некоторые из них имеют 0 изображений, а другие имеют 4 КБ, а среднее значение составляет 432. При обработке данных я получил эту ошибку:

Container killed by YARN for exceeding memory limits. 16.9 GB of 16 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.

в то время как memoryOverhead уже увеличена. Я чувствую, что происходят некоторые всплески, которые заставляют Yarn убивать мой контейнер, потому что этот всплеск выходит за указанные границы.

Итак, что я должен сделать, чтобы убедиться, что мои данные (Примерно)сбалансированы по разделам?

Моя идея состояла в том, чтопередел () будет работать, это вызывает перетасовку:

dataset = dataset.repartition(202092)

но я только что получил ту же ошибку, несмотря напрограммно-гидинструкция:

передел (numPartitions)

Перетасуйте данные в RDD случайным образом, чтобы создать больше или меньше разделов исбалансировать это через них, Это всегда перетасовывает все данные по сети.

Проверьте мой игрушечный пример, хотя:

data = sc.parallelize([0,1,2], 3).mapPartitions(lambda x: range((x.next() + 1) * 1000))
d = data.glom().collect()
len(d[0])     # 1000
len(d[1])     # 2000
len(d[2])     # 3000
repartitioned_data = data.repartition(3)
re_d = repartitioned_data.glom().collect()
len(re_d[0])  # 1854
len(re_d[1])  # 1754
len(re_d[2])  # 2392
repartitioned_data = data.repartition(6)
re_d = repartitioned_data.glom().collect()
len(re_d[0])  # 422
len(re_d[1])  # 845
len(re_d[2])  # 1643
len(re_d[3])  # 1332
len(re_d[4])  # 1547
len(re_d[5])  # 211
repartitioned_data = data.repartition(12)
re_d = repartitioned_data.glom().collect()
len(re_d[0])  # 132
len(re_d[1])  # 265
len(re_d[2])  # 530
len(re_d[3])  # 1060
len(re_d[4])  # 1025
len(re_d[5])  # 145
len(re_d[6])  # 290
len(re_d[7])  # 580
len(re_d[8])  # 1113
len(re_d[9])  # 272
len(re_d[10]) # 522
len(re_d[11]) # 66