我有一个luigipython 任务，其中包括一些 pyspark 库。现在我想用spark-submit在mesos上提交这个任务。我应该做什么来运行它？下面是我的代码框架：
from pyspark.sql import functions as F
from pyspark import SparkContext
class myClass(SparkSubmitTask):
# date = luigi.DateParameter()
def __init__(self, date):
self.date = date # date is datetime.date.today().isoformat()
def output(self):
def input(self):
def run(self):
# Some functions are using pyspark libs
if __name__ == "__main__":
luigi.run()
如果没有 luigi，我将通过以下命令行提交此任务：
/opt/spark/bin/spark-submit --master mesos://host:port --deploy-mode cluster --total-executor-cores 1 --driver-cores 1 --executor-memory 1G --driver-memory 1G my_module.py
现在的问题是我如何 Spark 提交包含 luigi 命令行的 luigi 任务，例如：
luigi --module my_module myClass --local-scheduler --date 2016-01
还有一个问题是，如果 my_module.py 有一项需要首先完成的任务，我是否需要为它做更多的事情，或者只是设置与当前命令行相同的值？
我真的很感谢对此的任何提示或建议。非常感谢。
Luigi 有一些模板任务。其中之一称为 PySparkTask。
您可以从此类继承并覆盖属性：
https://github.com/spotify/luigi/blob/master/luigi/contrib/spark.py https://github.com/spotify/luigi/blob/master/luigi/contrib/spark.py.
我还没有测试过，但根据我对 luigi 的经验，我会尝试这个：
import my_module
class MyPySparkTask(PySparkTask):
date = luigi.DateParameter()
@property
def name(self):
return self.__class__.__name__
@property
def master(self):
return 'mesos://host:port'
@property
def deploy_mode(self):
return 'cluster'
@property
def total_executor_cores(self):
return 1
@property
def driver_cores(self):
return 1
@property
def executor-memory(self):
return 1G
@property
def driver-memory(self):
return 1G
def main(self, sc, *args):
my_module.run(sc)
def self.app_options():
return [date]
然后你可以使用以下命令运行它：
luigi --module task_module MyPySparkTask --local-scheduler --date 2016-01
还有一个选项可以在 client.cfg 文件中设置属性，以使它们成为其他 PySparkTasks 的默认值：
[spark]
master: mesos://host:port
deploy_mode: cluster
total_executor_cores: 1
driver_cores: 1
executor-memory: 1G
driver-memory: 1G
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
