使用Luigi python运行Hadoop jar [英] Running Hadoop jar using Luigi python
问题描述
我需要使用Python中的 Luigi 来运行Hadoop jar作业。我搜索并找到了在Luigi中编写mapper和reducer的例子,但没有直接运行Hadoop jar的例子。
我需要运行直接编译的Hadoop jar。我该怎么做?
您需要使用 luigi.contrib.hadoop_jar
包( code )。
尤其需要扩展 HadoopJarJobTask
。例如,如下所示:
from luigi.contrib.hadoop_jar import HadoopJarJobTask
from luigi.contrib.hdfs.target import HdfsTarget
$ b $ 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文件:
导入LocalTarget
导入子流程
导入os
类BuildJobTask(luigi.Ta sk):
def输出(self):
返回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):
返回'com.ololo.HadoopJob'
def args(self):
return ['--param1','1','--param2','2']
def需要(个体):
返回BuildJobTask()
I need to run a Hadoop jar job using Luigi from python. I searched and found examples of writing mapper and reducer in Luigi but nothing to directly run a Hadoop jar.
I need to run a Hadoop jar compiled directly. How can I do it?
You need to use the luigi.contrib.hadoop_jar
package (code).
In particular, you need to extend HadoopJarJobTask
. For example, like that:
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']
You can also include building a jar file with maven to the workflow:
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()
这篇关于使用Luigi python运行Hadoop jar的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!