这是一个很好的问题。为了回答这个问题,我将使用PySpark wordcount 示例。
在这种情况下,我创建了两个文件,一个称为test.py
我要执行的文件,另一个称为wordcount.py.zip
zip,其中包含旨在模仿我要调用的模块的修改文件。 wordcount.py
我的test.py
文件如下所示:
import wordcount
import sys
if __name__ == "__main__":
wordcount.wctest(sys.argv[1])
我修改了wordcount.py
文件以消除主要方法并添加命名方法:
...
from pyspark import SparkContext
...
def wctest(path):
sc = SparkContext(appName="PythonWordCount")
...
我可以使用以下命令在Dataproc上调用整个过程:gcloud
gcloud beta dataproc jobs submit pyspark --cluster <cluster-name> \
--py-files gs://<bucket>/wordcount.py.zip gs://<bucket>/test.py \
gs://<bucket>/input/input.txt
在此示例<bucket>
中,是我的存储桶的名称(或路径),<cluster-name>
也是我的 Dataproc 集群的名称。