Блог

Я запускаю PySpark на ноутбуке Jupyter (на AWS EMR) и сталкиваюсь с какой-то странной ситуацией.

У меня была прекрасно работающая записная книжка, которая выполняла некоторые манипуляции с rdds, но я решил реорганизовать код в модули python, чтобы сделать его более читабельным. Я добавляю модули с помощью «sc.addPyFile(«myCodePackage.zip»)» команда.

После реорганизации части кода я столкнулся с проблемой — все действия, выполняемые на rdds из импортированных файлов, приводят к зависанию PySpark — это происходит для каждого действия, выполняемого из какой-либо импортированной функции (из определенного модуля python). Когда я извлекаю объявление функции в какую-либо ячейку в записной книжке (ничего не меняя в коде функции) и вызываю ее, она работает нормально.

Пример такой функции —

 import gzip
import json

def parse_items_from_compressed_json(json_str):
    items = json.loads(gzip.decompress(json_str))
    return items

def example(gzip_path):
    return sc.binaryFiles(gzip_path).flatMap(lambda x:     
    parse_items_from_compressed_json(x[1]))
 

Когда этот код находится в ячейке в записной книжке — он работает идеально!

При вызове его из импортированного файла он зависает..

Спасибо за любую помощь.