Блог

Я довольно новичок в Spark и распределенных вычислениях. Все очень просто для загрузки файла csv или текстового файла, который может поместиться в память вашего драйвера. Но здесь у меня реальный сценарий, и мне трудно понять подход.

Я пытаюсь получить доступ к примерно 500 Г данных в S3, они состоят из Zip-файлов. поскольку это zip-файлы, я использую ZipFileInputFormat, как описано здесь . Это гарантирует, что файлы не разделяются по разделам.

Вот мой код val sc = новый SparkContext(conf)

 val inputLocation = args(0) 
val emailData = sc.newAPIHadoopFile(inputLocation, classOf[ZipFileInputFormat], classOf[Text], classOf[BytesWritable]);

val filesRDD = emailData.filter(_._1.toString().endsWith(".txt")).map( x => new String(x._2.getBytes))
  

Это нормально работает при вводе нескольких 100 МБ. но как только он пересекает лимит памяти моего кластера, я получаю проблему OutOfMemory.

как правильно подойти к этой проблеме? — должен ли я создавать RDD для каждого zip-файла и сохранять выходные данные в файл, а затем загружать все выходные данные в отдельный RDD позже? — Есть ли способ загрузить базовый каталог в контекст Spark и разделить

У меня есть HDP-кластер с 5 узлами и ведущим, каждый из которых имеет 15 ГБ памяти.

Любые ответы / указатели / ссылки высоко ценятся

zip файлы не разделяются, поэтому обработка отдельных файлов не принесет вам никакой пользы. Если вы не хотите его масштабировать, вам следует полностью избегать их или, по крайней мере, жестко ограничивать размер архивов.