将snappy压缩文件加载到Elastic MapReduce中

Question

我在S3中有一堆snappy压缩的服务器日志,我需要在Elastic MapReduce上使用流处理它们.我如何告诉亚马逊和Hadoop日志已被压缩(在它们被拉入HFS之前!),以便在发送到流式映射器脚本之前将它们解压缩？

我能找到的唯一文档是:http://docs.aws.amazon.com/ElasticMapReduce/latest/DeveloperGuide/HadoopDataCompression.html#emr-using-snappy ,它似乎是指中间压缩,而不是文件当他们到达HFS时被压缩.

顺便说一句,我主要是在python工作,所以如果你有一个boto解决方案的奖励积分!

Answer 1

答案是,"它无法完成." 至少,不适用于将hadoop流应用于源自hadoop之外的snappy压缩文件的特定情况.

我(彻底!)探索了两个主要选项来得出这个结论:(1)尝试使用hadoop内置的snappy压缩,如高度咖啡因所暗示,或(2)编写我自己的流模块来使用和解压缩snappy文件.

对于选项(1),似乎hadoop在使用snappy压缩文件时为文件添加了一些标记.由于我的文件是使用snappy在hadoop外部压缩的,因此hadoop的内置编解码器无法解压缩文件.

此问题的一个症状是堆空间错误:

2013-04-03 20:14:49,739 FATAL org.apache.hadoop.mapred.Child (main): Error running child : java.lang.OutOfMemoryError: Java heap space
    at org.apache.hadoop.io.compress.BlockDecompressorStream.getCompressedData(BlockDecompressorStream.java:102)
    at org.apache.hadoop.io.compress.BlockDecompressorStream.decompress(BlockDecompressorStream.java:82)
    at org.apache.hadoop.io.compress.DecompressorStream.read(DecompressorStream.java:76)
    at java.io.InputStream.read(InputStream.java:85)
    ...

当我切换到一个更大的实例并启动mapred.child.java.opts设置时,我收到一个新错误:

java.io.IOException: IO error in map input file s3n://my-bucket/my-file.snappy

Hadoop的snappy编解码器无法与外部生成的文件一起使用.

对于选项(2),问题是hadoop流不区分\n,\ r和\ r \n换行符.由于snappy压缩最终会在压缩文件中传播这些字节代码,因此这是致命的.这是我的错误跟踪:

2013-04-03 22:29:50,194 WARN org.apache.hadoop.mapred.Child (main): Error running child
java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1
    at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:372)
    at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:586)
    at org.apache.hadoop.streaming.PipeMapper.close(PipeMapper.java:135)
    at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:57)
    ...

通过对hadoop的Java类的一些工作(例如,见这里),我们可以解决\ r vs \n \n问题.但正如我最初所说,我的目标是在hadoop流模块中构建,而不涉及Java.有了这个约束,似乎没有办法解决这个问题.

最后,我回到了生成这个集群正在消耗的文件的人,并说服他们切换到gzip或lzo.

PS - 在选项(2)上,我玩弄了不同角色的分割记录(例如textinputformat.record.delimiter = X),但它感觉非常hacky并且无论如何都不起作用.

PPS - 另一种解决方法是编写脚本从S3下载文件,解压缩它们,然后运行-copyFromLocal将它们拉入HDFS.在计算上,这没有任何问题,但从工作流的角度来看,它会引入各种麻烦.