我想在我的mapreduce中包含已编译的proto类，我认为最简单的方法是将所需的python文件压缩为tar.gz并上传到s3。
我遇到了这个stackoverflow答案，并按照以下步骤创建了存档文件。然后我用 -cacheArchive 参数将文件分发到群集。
使用boto启动emr作业我的配置如下：

Steps=[{
    'Name': '...',
    'ActionOnFailure': 'TERMINATE_CLUSTER',
    'HadoopJarStep': {
        'Jar': 'command-runner.jar',
        'Args':
            ['hadoop-streaming',
             '-files', 's3://myBucket/mapper.py,'
                       's3://myBucket/reducer.py',
             '-mapper', 'mapper.py',
             '-input', 's3://myBucket/input/',
             '-output', 's3://myBucket/output',
             '-reducer', 'reducer.py',
             '-cacheArchive', 's3://myBucket/required.tgz#required']
    }
}]

Map器.py

# !/usr/bin/env python

import sys

sys.path.append('./required')
from primitives_pb2 import Variant

for line in sys.stdin:
    # DO THINGS

我得到以下错误时，电子病历。
错误：java.lang.runtimeexception:pipemapred.waitoutputthreads（）：org.apache.hadoop.streaming.pipemapred.waitoutputthreads（pipemapred）处的子进程失败，代码为1。java:332)在org.apache.hadoop.streaming.pipemapred.mapredfinished（pipemapred。java:545)在org.apache.hadoop.streaming.pipemapper.close（pipemapper。java:130)在org.apache.hadoop.mapred.maprunner.run（maprunner。java:81)在org.apache.hadoop.streaming.pipemaprunner.run（pipemaprunner。java:34)在org.apache.hadoop.mapred.maptask.runoldmapper（maptask。java:452)在org.apache.hadoop.mapred.maptask.run（maptask。java:344)在org.apache.hadoop.mapred.yarnchild$2.run（yarnchild。java:171)在javax.security.auth.subject.doas（主题）中的java.security.accesscontroller.doprivileged（本机方法）。java:415)在org.apache.hadoop.security.usergroupinformation.doas（usergroupinformation。java:1628)在org.apache.hadoop.mapred.yarnchild.main（yarnchild。java:166)
当在本地运行mapreduce时，它似乎工作得很好，所以我也不认为我的编写方式有错误 mapper.py 或者 reducer.py .