使用 MRJob 将作业提交到 EMR 集群

Posted 2023-04-18

【中文标题】使用 MRJob 将作业提交到 EMR 集群

【英文标题】：Submit jobs to EMR cluster using MRJob

【发布时间】：2016-04-26 19:50:57

【问题描述】：

MRJob 等到每个作业完成后再将控制权交还给用户。我将一个大型 EMR 步骤分解为多个较小的步骤，并希望一次性提交它们。

文档讨论了programmatically submitting tasks，但示例代码也等待作业完成（因为他们调用了blocks until the job is complete 的runner.run() 命令）。

此外，EMR 有 256 个活动作业的限制，但是，我们如何填充这 256 个作业，而不是循环并在附加的控制台上获取输出。

【参考方案1】：

经过几天的尝试，以下是我能想到的最好的。

我最初的尝试是，当我意识到当终端分离时提交的作业不会被剔除时，是（在 bash 脚本中）提交并终止作业。然而，这并没有很好地工作，因为 AWS 限制了对 EMR 的调用，因此一些作业在提交之前就被终止了。

当前最佳解决方案

from jobs import MyMRJob
import logging

logging.basicConfig(
    level=logging.INFO,
    format = '%(asctime)-15s %(levelname)-8s %(message)s',
)
log = logging.getLogger('submitjobs')

def main():
    cluster_id="x-MXMXMX"
    log.info('Cluster: %s', cluster_id)
    for i in range(10):
        n = '%04d' % i
        log.info('Adding job: %s', n)
        mr_job = MyMRJob(args=[
            '-r', 'emr',
            '--conf-path', 'mrjob.conf',
            '--no-output',
            '--output-dir', 's3://mybucket/mrjob/%s' % n,
            '--cluster-id', cluster_id,
            'input/file.%s' % n
    ])
    runner = mr_job.make_runner()
    # the following is the secret sauce, submits the job and returns
    # it is a private method though, so may be changed without notice
    runner._launch()

if __name__ == '__main__':
    main()