你需要使用 luigi.contrib.hadoop_jar 包（代码）。
特别是，你需要扩展 HadoopJarJobTask . 例如：

from luigi.contrib.hadoop_jar import HadoopJarJobTask
from luigi.contrib.hdfs.target import HdfsTarget
class TextExtractorTask(HadoopJarJobTask):
    def output(self):
        return HdfsTarget('data/processed/')
    def jar(self):
        return 'jobfile.jar'
    def main(self):
        return 'com.ololo.HadoopJob'
    def args(self):
        return ['--param1', '1', '--param2', '2']

您还可以将使用maven构建jar文件包括到工作流中：

import luigi
from luigi.contrib.hadoop_jar import HadoopJarJobTask
from luigi.contrib.hdfs.target import HdfsTarget
from luigi.file import LocalTarget
import subprocess
import os
class BuildJobTask(luigi.Task):
    def output(self):
        return LocalTarget('target/jobfile.jar')
    def run(self):
        subprocess.call(['mvn', 'clean', 'package', '-DskipTests'])
class YourHadoopTask(HadoopJarJobTask):
    def output(self):
        return HdfsTarget('data/processed/')
    def jar(self):
        return self.input().fn
    def main(self):
        return 'com.ololo.HadoopJob'
    def args(self):
        return ['--param1', '1', '--param2', '2']
    def requires(self):
        return BuildJobTask()