Загрузка файлов из Google Storage с использованием Spark (Python) и Dataproc

Question

У меня есть приложение, которое распараллеливает выполнение объектов Python, которые обрабатывают данные, которые будут загружаться из Google Storage (мой проект). Кластер создается с помощью Google Dataproc. Проблема в том, что данные никогда не загружаются! Я написал тестовую программу, чтобы попытаться понять проблему. я написал следующую функцию для копирования файлов из ведра и увидеть, если создание файлов на рабочих делает работу:

from subprocess import call 
from os.path import join 

def copyDataFromBucket(filename,remoteFolder,localFolder): 
    call(["gsutil","-m","cp",join(remoteFolder,filename),localFolder] 

def execTouch(filename,localFolder): 
    call(["touch",join(localFolder,"touched_"+filename)]) 

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

# ... 
filesRDD = sc.parallelize(fileList) 
filesRDD.foreach(lambda myFile: copyDataFromBucket(myFile,remoteBucketFolder,'/tmp/output') 
filesRDD.foreach(lambda myFile: execTouch(myFile,'/tmp/output') 
# ... 

Функция работает execTouch (я могу видеть файлы на каждого работника), но функцию copyDataFromBucket ничего не делает.

Так что я делаю неправильно?

Answer 1

Проблема была явно контекстом Spark. Замена вызова «GSUtil» вызовом «Hadoop фс» решает эту проблему:

from subprocess import call 
from os.path import join 

def copyDataFromBucket(filename,remoteFolder,localFolder): 
    call(["hadoop","fs","-copyToLocal",join(remoteFolder,filename),localFolder] 

Я также сделал тест, чтобы отправить данные в ведро. Нужно заменить «-copyToLocal» на «-copyFromLocal»