如何将avro文件作为MRJob工作输入？

Question

我需要把avro文件作为输入，放到一个mrjob的hadoop任务里。但是我找不到相关的文档，除非我给hadoop的流处理jar加一些额外的命令。不过这样会让开发变得复杂，因为我一直在用inline这个运行方式在本地测试。

请问能不能用inline这个运行方式来读取avro文件呢？

Answer 1

正如Chiron所说，你需要指定Hadoop的输入格式。这可以通过在MRJob中设置HADOOP_INPUT_FORMAT选项来完成。

from mrjob.job import MRJob
from mrjob.protocol import JSONProtocol
class MRAvro(MRJob):
    # Converts each AVRO record into one JSON record per line 
    HADOOP_INPUT_FORMAT = 'org.apache.avro.mapred.AvroAsTextInputFormat'
    # Reads each JSON line into 
    INPUT_PROTOCOL = JSONProtocol

    def mapper(self, avro_record, _):
        # TODO

    def reducer(self, key, values):
        # TODO

在你的配置中，你需要确保AvroAsTextInputFormat的.jar文件在集群上是可用的；从版本0.5.3开始，你可以在命令行使用--libjar，或者在mrjob配置文件中配置libjars（在0.5.3版本发布之前，可以查看关于--libjar的讨论，链接在这里）。

我不知道有没有简单的方法可以将本地测试与AVRO结合起来（本地运行时会忽略HADOOP_INPUT_FORMAT）。一种解决方案是使用Apache avro-tools的tojson方法将你的测试数据转换。

java -jar avro-tools-1.8.1.jar test_data.avro > test_data.json

另外，你也可以使用avro或fastavro库在Python中编写自己的函数，以准备数据进行本地执行。

Answer 2

你需要告诉Hadoop你的“输入格式”是什么，这样它才能正确处理你的Hadoop任务：

hadoop jar hadoop-streaming.jar 
  ;; other params go here
  -inputformat org.apache.avro.mapred.AvroAsTextInputFormat

不过我不太确定你是怎么运行MRJobs的。如果你使用的是普通的Hadoop，我之前的解决方案是有效的。