python - 如何使用 spark-submit 和 pyspark 运行 luigi 任务

Question

我有一个 luigi python 任务，其中包括一些 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

现在的问题是我如何激发提交包含 luigi 命令行的 luigi 任务，例如:

luigi --module my_module myClass --local-scheduler --date 2016-01

还有一个问题是，如果 my_module.py 有一个必须首先完成的任务，我是否需要为它做更多的事情，或者只是设置与当前命令行相同？

我非常感谢对此的任何提示或建议。非常感谢。

Answer 1

Luigi 有一些模板任务。其中之一称为 PySparkTask。 您可以继承此类并覆盖属性:

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