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