针对Apache Flink的BZip2压缩输入

我有一个使用bzip2压缩的wikipedia转储（从http://dumps.wikimedia.org/enwiki/下载），但我不想解压缩它：我想在进行解压缩的同时处理它。针对Apache Flink的BZip2压缩输入

我知道可以用普通的Java来完成它（例如见Java - Read BZ2 file and uncompress/parse on the fly），但是我想知道它在Apache Flink中是如何做到的？我可能需要的是类似于https://github.com/whym/wikihadoop但对于Flink而不是Hadoop。

来源

2015-04-03 Alexey Grigorev

有可能在阿帕奇弗林克以下格式读取压缩文件：

org.apache.hadoop.io.compress.BZip2Codec 
org.apache.hadoop.io.compress.DefaultCodec 
org.apache.hadoop.io.compress.DeflateCodec 
org.apache.hadoop.io.compress.GzipCodec 
org.apache.hadoop.io.compress.Lz4Codec 
org.apache.hadoop.io.compress.SnappyCodec

你可以从包名看，弗林克这是否使用Hadoop的InputFormats。这是阅读使用弗林克的斯卡拉API GZ文件的示例：（你至少需要弗林克0.8.1）

def main(args: Array[String]) { 

    val env = ExecutionEnvironment.getExecutionEnvironment 
    val job = new JobConf() 
    val hadoopInput = new TextInputFormat() 
    FileInputFormat.addInputPath(job, new Path("/home/robert/Downloads/cawiki-20140407-all-titles.gz")) 
    val lines = env.createHadoopInput(hadoopInput, classOf[LongWritable], classOf[Text], job) 

    lines.print 

    env.execute("Read gz files") 
}

阿帕奇弗林克只有内建支持.deflate文件。添加对更多压缩编解码器的支持很容易，但尚未完成。

将HadoopInputFormats与Flink结合使用不会导致任何性能损失。 Flink拥有对Hadoop的Writable类型的内建序列化支持。

来源

2015-04-03 11:49:20

针对Apache Flink的BZip2压缩输入

回答

相关问题