在hadoop中使用可变长度/非分隔二进制文件进行拆分

Question

我刚刚开始开发一个基于hadoop的开放式街道地图数据的ingester。有几种格式--但我的目标是基于protocolbuffer的格式（注意--它不是纯粹的pb）。

在我看来，与以自定义记录读取器/输入格式处理可变长度编码相反，将文件预拆分为序列文件将更有效，但希望进行健全性检查。

该格式在PBF格式描述中有更详细的描述，但基本上它是[BlobHeader,blob]块的集合。

有一个Blob头

message BlobHeader {
   required string type = 1;
   optional bytes indexdata = 2;
   required int32 datasize = 3;
 }

 message Blob {
   optional bytes raw = 1; // No compression
   optional int32 raw_size = 2; // Only set when compressed, to the uncompressed size
   optional bytes zlib_data = 3;
   // optional bytes lzma_data = 4; // PROPOSED.
   // optional bytes OBSOLETE_bzip2_data = 5; // Deprecated.
 }

有没有更好的方法来处理这个问题--或者如果没有，对这两个建议的想法？

Answer 1

嗯，我使用getSplits方法解析二进制文件--因为我跳过了99%以上的html" target="_blank">数据，所以它非常快（对于planet-osm 22GB的世界文件来说大约20秒）。下面是getSplits方法，如果有人遇到困难的话。

@Override
public List<InputSplit> getSplits(JobContext context){
    List<InputSplit> splits = new ArrayList<InputSplit>();
    FileSystem fs = null;
    Path file = OSMPBFInputFormat.getInputPaths(context)[0]; 
    FSDataInputStream in = null;
    try {
        fs = FileSystem.get(context.getConfiguration());
        in = fs.open(file);
        long pos = 0;
        while (in.available() > 0){
            int len = in.readInt(); 
            byte[] blobHeader = new byte[len]; 
            in.read(blobHeader);
            BlobHeader h = BlobHeader.parseFrom(blobHeader);
            FileSplit split = new FileSplit(file, pos,len + h.getDatasize(), new String[] {});
            splits.add(split);
            pos += 4;
            pos += len;
            pos += h.getDatasize();
            in.skip(h.getDatasize());
        }
    } catch (IOException e) {
        sLogger.error(e.getLocalizedMessage());
    } finally {
        if (in != null) {try {in.close();}catch(Exception e){}};
        if (fs != null) {try {fs.close();}catch(Exception e){}};
    }
    return splits;
}

到目前为止工作得很好--尽管我还没有得到真实的输出结果。它比将pbf复制到hdfs更快，在单个映射器中转换为序列，然后摄入（复制时间占主导地位）。它还比将外部程序复制到hdfs中的序列文件，然后针对hdfs运行映射器（为后者编写脚本）快20%。所以这里没有抱怨。

请注意，这将为每个块生成一个映射器--对于planet world文件来说，这是23K映射器。我实际上是在每个拆分捆绑多个块--在将拆分添加到集合之前，只需循环x次。

对于BlobHeader，我只是从上面的OSM wiki链接编译了protobuf.proto文件。如果需要，还可以从OSM-binary类中提取预生成的代码--maven片段为：

<dependency>
    <groupId>org.openstreetmap.osmosis</groupId>
    <artifactId>osmosis-osm-binary</artifactId>
    <version>0.43-RELEASE</version>
</dependency>