Как вы перебираете отдельные значения столбца в большом фрейме данных Pyspark? .distinct (). collect () вызывает предупреждение о большой задаче

Question

Я пытаюсь перебрать все различные значения в столбце большого фрейма данных Pyspark. Когда я пытаюсь сделать это, используя .distinct().collect(), появляется предупреждение «слишком большая задача», даже если есть только два разных значения. Предупреждающее сообщение:

20/01/13 20:39:01 WARN TaskSetManager: Stage 0 contains a task of very large size (201 KB). The maximum recommended task size is 100 KB.

Вот пример кода:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('Basics').getOrCreate()
length = 200000

data = spark.createDataFrame([[float(0) for x in range(3)] for x in range(length)], ['a', 'b', 'c'])

data.select("a").distinct().collect()
# This code produces this warning

Как выполнить итерацию по отдельным значениям в столбце большого фрейма данных Pyspark без проблем с памятью?

Answer 1

Как вы уже знаете, .collect() не лучшая практика. Потому что это действие, которое передает все данные от исполнителей к водителю. Проблема в том, что когда у вас большой набор данных, исполнители Spark отправили в драйвер большой объем сериализованных данных, а затем выполняют сбор из 2 строк. Вы также можете взглянуть на TaskSetManager , который выдает предупреждение.

На высоком уровне, решением проблемы может быть обмен памятью с диском. Вы можете записать свой фрейм данных с разными значениями в одном csv, а затем снова и снова читайте его построчно с Python или Pandas*:

data.select("a").distinct().coalesce(1).write.csv("temp.csv")
# Specifically, it's a directory with one csv.

С этим решением у вас не возникнет проблем с памятью.

* Существует множество решений о том, как читать большой CSV с Python или Pandas.