我正在尝试使用 mrjob 在 EMR 上运行 hadoop,但不知道如何设置日志记录(用户在 map/reduce 步骤中生成的日志),因此我将能够在集群终止后访问它们。
我尝试使用 logging
模块、print
和 sys.stderr.write()
来设置日志记录,但到目前为止还没有成功。对我有用的唯一选择是将日志写入文件,然后 SSH 机器并读取它,但它很麻烦。我希望我的日志转到 stderr/stdout/syslog 并自动收集到 S3,这样我就可以在集群终止后查看它们。
这是带有日志记录的 word_freq 示例:
"""The classic MapReduce job: count the frequency of words.
"""
from mrjob.job import MRJob
import re
import logging
import logging.handlers
import sys
WORD_RE = re.compile(r"[\w']+")
class MRWordFreqCount(MRJob):
def mapper_init(self):
self.logger = logging.getLogger()
self.logger.setLevel(logging.INFO)
self.logger.addHandler(logging.FileHandler("/tmp/mr.log"))
self.logger.addHandler(logging.StreamHandler())
self.logger.addHandler(logging.StreamHandler(sys.stdout))
self.logger.addHandler(logging.handlers.SysLogHandler())
def mapper(self, _, line):
self.logger.info("Test logging: %s", line)
sys.stderr.write("Test stderr: %s\n" % line)
print "Test print: %s" % line
for word in WORD_RE.findall(line):
yield (word.lower(), 1)
def combiner(self, word, counts):
yield (word, sum(counts))
def reducer(self, word, counts):
yield (word, sum(counts))
if __name__ == '__main__':
MRWordFreqCount.run()
最佳答案
在所有选项中,唯一真正有效的是使用直接写入的 stderr (sys.stderr.write
) 或使用带有 StreamHandler 的记录器到 stderr。
作业完成后(成功或出错)可以从以下位置检索日志:
[s3_log_uri]/[jobflow-id]/task-attempts/[job-id]/[attempt-id]/stderr
请务必将日志保存在您的 runners.emr.cleanup
配置中。
关于python - mrjob:在 EMR 上设置日志记录,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/26123152/