3

我正在使用 MRjob 在我们的 HBase 实例上运行 Hadoop Streaming 作业。对于我的生活，我无法弄清楚如何将参数传递给我的减速器。我有两个参数要在运行作业时传递给减速器：startDate 和 endDate。这是我目前的减速器的样子：

def reducer(self, groupId, meterList):
    """
    Print bucket.
    """
    sys.stderr.write("Working on group = " + str(groupId) + "\n")
    #print "Opening connection..."
    conn = open_connection(hostname)
    #print "Getting table..."
    table = get_table(conn, tableName)

    compositeDf = DataFrame()

    for meterId in meterList:
        sys.stderr.write("Querying: " + str(meterId) + "\n")
        df = extract_meter_data(table, meterId, startDate, endDate)

我似乎无法将 startDate 和 endDate 作为参数传递给我的减速器。我可以获取参数的唯一方法是通过类顶部的全局变量。

startDate = datetime.datetime(2012, 6, 10)
endDate = datetime.datetime(2012, 6, 11)

class MRDataQuality(MRJob):
    """
    MapReduce job that does a data quality check on the meter data in HBase.
    """

但这很脏。我想通过调用工作来传递它。我试过很多方法。将其设置为实例变量，将其设置为静态类变量，为 MRDataQualityJob 创建重载构造函数......似乎没有任何效果。我以编程方式从我的顶级脚本中调用它，如下所示：

if args.hadoop:
    mrdq_job = MRDataQuality(args=['-r', 'hadoop', '--conf-path', 'mrjob.conf', '--jobconf', 'mapred.reduce.tasks=42', meterFile])
else:
    mrdq_job = MRDataQuality(args=[meterFile])

with mrdq_job.make_runner() as runner:
    runner.run()

无论我对 mrdq_job 实例做什么，似乎 runner.run() 正在使用该类的新实例，该实例没有定义实例或静态变量。我怎样才能将我的参数传递给减速器？？？？我可以通过传递一个字符串在常规 Hadoop Streaming 中做到这一点：“--reducer reducer.py arg1 arg2”。MRjob有什么等价物吗？

4

2 回答 2

3

将您的参数传递给作业配置，然后使用 get_jobconf_value 读取它们怎么样？

像这样的东西：

from mrjob.compat import get_jobconf_value

class MRDataQuality(MRJob):

  def reducer(self, groupId, meterList):
    ...
    startDate = get_jobconf_value("my.job.settings.startdate")
    endDate = get_jobconf_value("my.job.settings.enddate")

    for meterId in meterList:
      sys.stderr.write("Querying: " + str(meterId) + "\n")
      df = extract_meter_data(table, meterId, startDate, endDate)

然后像上面一样在代码中设置参数

mrdq_job = MRDataQuality(args=['-r', 'hadoop', '--conf-path', 'mrjob.conf', '--jobconf', 'mapred.reduce.tasks=42', '--jobconf', 'my.job.settings.startdate=2013-06-10', '--jobconf', 'my.job.settings.enddate=2013-06-11', meterFile])
于 2013-08-23T15:40:41.163 回答
1

将你的参数传递给作业配置，然后在 reducer_init 中使用 get_jobconf_value 读取它们怎么样？这样，您只需读取一次参数。

像这样的东西：

from mrjob.compat import get_jobconf_value

class MRDataQuality(MRJob):

  def reducer_init(self):
    ...
    self.startDate = get_jobconf_value("my.job.settings.startdate")
    self.endDate = get_jobconf_value("my.job.settings.enddate")

  def reducer(self, groupId, meterList):
    for meterId in meterList:
      sys.stderr.write("Querying: " + str(meterId) + "\n")
      df = extract_meter_data(table, meterId, self.startDate, self.endDate)

然后像上面一样在代码中设置参数

mrdq_job = MRDataQuality(args=['-r', 'hadoop', '--conf-path', 'mrjob.conf', '--jobconf', 'mapred.reduce.tasks=42', '--jobconf', 'my.job.settings.startdate=2013-06-10', '--jobconf', 'my.job.settings.enddate=2013-06-11', meterFile])
于 2013-10-22T19:16:05.620 回答