用mrjob处理LZO序列文件

Question

我正在使用马约伯编写一个任务，使用Google数据计算各种统计数据：https://aws.amazon.com/datasets/8172056142375670

我使用标签分隔文本中数据的未压缩子集在本地开发和测试了我的脚本。一旦我试图运行该作业，就会出现以下错误：

Traceback (most recent call last):
  File "ngram_counts.py", line 74, in <module>
    MRNGramCounts.run()
  File "/usr/lib/python2.6/dist-packages/mrjob/job.py", line 500, in run
    mr_job.execute()
  File "/usr/lib/python2.6/dist-packages/mrjob/job.py", line 509, in execute
    self.run_mapper(self.options.step_num)
  File "/usr/lib/python2.6/dist-packages/mrjob/job.py", line 574, in run_mapper
    for out_key, out_value in mapper(key, value) or ():
  File "ngram_counts.py", line 51, in mapper
    (ngram, year, _mc, _pc, _vc) = line.split('\t')
ValueError: need more than 2 values to unpack
(while reading from s3://datasets.elasticmapreduce/ngrams/books/20090715/eng-1M/5gram/data)

这大概是因为公共数据集的压缩方案(来自上面的URL链接)：

我们将数据集存储在亚马逊S3中的单个对象中。该文件采用顺序文件格式，具有块级LZO压缩。序列文件键是作为LongWritable存储的数据集的行号，值是作为TextWritable存储的原始数据。

有关于如何设置可以处理这些文件的工作流的指导吗？我已经找遍了所有的窍门，但没有发现任何有用的东西.

(我是mrjob和Hadoop的相对n00b。)

Answer 1

我终于弄明白了。看起来EMR负责LZO压缩，但是对于序列文件格式，您需要将以下HADOOP_INPUT_FORMAT字段添加到您的MRJob类中：

class MyMRJob(MRJob):

    HADOOP_INPUT_FORMAT = 'org.apache.hadoop.mapred.SequenceFileAsTextInputFormat'

    def mapper(self, _, line):
        # mapper code...

    def reducer(self, key, value):
        # reducer code...

还有另一个问题(引用由AWS托管的Google NGrams页面)：

序列文件键是作为LongWritable存储的数据集的行号，值是作为TextWritable存储的原始数据。

这意味着每一行都被加上了超长+ TAB，所以在mapper方法中所做的任何行解析都需要考虑前面的信息。