Как загрузить огромный CSV в DataFrame Pyspark?

Question

Я пытаюсь загрузить огромный набор данных c генома (2504 строки и 14848614 столбцов) в PySpark DataFrame, но безуспешно. Я получаю java.lang.OutOfMemoryError: Java heap space. Я думал, что основной идеей использования искры была именно независимость памяти ... (я новичок ie на ней. Пожалуйста, потерпите меня :)

Это мой код:

from pyspark.sql import SparkSession

spark = SparkSession.builder.config("spark.driver.memory", "6G").getOrCreate()

file_location = "1kGp3_chr3_6_10.raw"
file_type = "csv"

infer_schema = "true"
first_row_is_header = "true"
delimiter = "\t"
max_cols = 15000000 # 14848614 variants loaded

data = spark.read.format(file_type) \
  .option("inferSchema", infer_schema) \
  .option("header", first_row_is_header) \
  .option("sep", delimiter) \
  .option("maxColumns", max_cols) \
  .load(file_location)

Я знаю, что мы можем установить StorageLevel, например, df.persist(StorageLevel.DISK_ONLY), но это возможно только после успешной загрузки файла в DataFrame, не так ли? (не уверен, что я что-то упустил)

Вот ошибка:

...
Py4JJavaError: An error occurred while calling o33.load.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 1, localhost, executor driver): java.lang.OutOfMemoryError: Java heap space
...

Спасибо!

---

РЕДАКТИРОВАТЬ / ОБНОВИТЬ: Я забыл Чтобы упомянуть размер CSV: 70G.

Вот еще одна попытка, которая привела к другой ошибке: я попытался с меньшим набором данных (2504 строки и 3992219 столбцов. Размер файла: 19G) и увеличил объем памяти до "spark.driver.memory", "12G". Примерно через 35 минут работы метода загрузки я получил:

Py4JJavaError: An error occurred while calling o33.load.
: org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 54 tasks (1033.1 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
    at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1889)

Answer 1

Ваша ошибка говорит о проблеме - у вас недостаточно памяти.

Значение при использовании pyspark - это не независимость памяти, а ее скорость, потому что (она использует оперативную память), возможность сохранять определенные данные или операции и возможность использовать несколько машин.

Итак, решения -

1) Если возможно, выделите больше оперативной памяти.

2) В зависимости от размера вашего файла CSV, вы можете или не сможете поместить его в память для ноутбук или рабочий стол. В этом случае вам может потребоваться поместить его в нечто вроде облачного экземпляра по соображениям скорости или стоимости. Даже там вы можете не найти машину, достаточно большую, чтобы вместить все это в память для одной машины (хотя, честно говоря, это было бы довольно большим, если учесть, что текущий максимум Amazon для одного экземпляра с оптимизированной памятью (u-24tb1.metal) равен 24 576 ГиБ.

И вот вы видите истинную силу pyspark: возможность загружать по-настоящему гигантские наборы данных в оперативную память и запускать ее на нескольких машинах.