Hadoop处理HDF文件

1、前言

HDF文件是遥感应用中一种常见的数据格式，由于其高度结构化的特点，笔者曾被如何使用Hadoop处理HDF文件这个问题困扰过相当长的一段时间。于是Google各种解决方案，但都没有找到一种理想的处理办法。也曾参考过HDFGroup官方发的一篇帖子（网址在这里），里面提供了使用Hadoop针对大、中、小HDF文件的处理思路。虽然根据他提供的解决办法，按图索骥，肯定能解决如何使用Hadoop处理HDF文件这个问题，但个人感觉方法偏复杂且需要对HDF的数据格式有较深的理解，实现起来不太容易。于是乎，笔者又继续寻找解决方案，终于发现了一种办法，下面将对该方法进行具体说明。

2、MapReduce主程序

这里主要使用到了netcdf的库进行hdf数据流的反序列化工作（netcdf库的下载地址）。与HDF官方提供的Java库不同，netcdf仅利用Java进行HDF文件的读写操作，且这个库支持多种科学数据，包括HDF4、HDF5等多种格式。而HDF的官方Java库中，底层实际仍是用C进行HDF文件的操作。

下面贴出MapReduce的Mapper函数代码：

package example;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.net.URI;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import ucar.ma2.ArrayShort;
import ucar.nc2.Dimension;
import ucar.nc2.Group;
import ucar.nc2.NetcdfFile;
import ucar.nc2.Variable;

public class ReadMapper extends
  Mapper<Text, BytesWritable, Text, BytesWritable> {

 public void map(Text key, BytesWritable value, Context context)
   throws IOException, InterruptedException { 
  String fileName = key.toString();
  NetcdfFile file = NetcdfFile.openInMemory("hdf4", value.get());
  Group dataGroup = (file.findGroup("MOD_Grid_monthly_1km_VI")).findGroup("Data_Fields");
  //读取到1_km_monthly_red_reflectance的变量
  Variable redVar = dataGroup.findVariable("1_km_monthly_red_reflectance");
  short[][] data = new short[1200][1200];
  if(dataGroup != null){   
   ArrayShort.D2 dataArray;
   //读取redVar中的影像数据
   dataArray = (ArrayShort.D2) redVar.read();
   List<Dimension> dimList = file.getDimensions();
   //获取影像的y方向像元个数
   Dimension ydim = dimList.get(0);
   //获取影像的x方向像元个数
   Dimension xdim = dimList.get(1);
   //遍历整个影像，读取出像元的值
   for(int i=0;i<xdim.getLength();i++){
    for(int j=0;j<ydim.getLength();j++){
     data[i][j] = dataArray.get(i, j);     
    }   
   }         
  } 
  System.out.print(file.getDetailInfo());
 }
}

注意程序中的NetcdfFile.openInMemory方法，该静态方法支持从byte[]中构造HDF文件，从而实现了HDF文件的反序列化操作。下面贴出主程序的示例代码：

package example;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;

import example.WholeFileInputFormat;

public class ReadMain {
 public boolean runJob(String[] args) throws IOException,
   ClassNotFoundException, InterruptedException {
  Configuration conf = new Configuration();
  // conf.set("mapred.job.tracker", Utils.JOBTRACKER);
  String rootPath= "/opt/hadoop-2.3.0/etc/hadoop";
  //String rootPath="/opt/hadoop-2.3.0/etc/hadoop/";
  conf.addResource(new Path(rootPath+"yarn-site.xml"));
  conf.addResource(new Path(rootPath+"core-site.xml"));
  conf.addResource(new Path(rootPath+"hdfs-site.xml"));
  conf.addResource(new Path(rootPath+"mapred-site.xml"));
  Job job = new Job(conf);

  job.setJobName("Job name:" + args[0]);
  job.setJarByClass(ReadMain.class);

  job.setMapperClass(ReadMapper.class);
  job.setMapOutputKeyClass(Text.class);
  job.setMapOutputValueClass(BytesWritable.class);
 
  job.setInputFormatClass(WholeFileInputFormat.class);
  job.setOutputFormatClass(NullOutputFormat.class);
  FileInputFormat.addInputPath(job, new Path(args[1]));
  FileOutputFormat.setOutputPath(job, new Path(args[2]));
  boolean flag = job.waitForCompletion(true);
  return flag;
 }

 public static void main(String[] args) throws ClassNotFoundException,
   IOException, InterruptedException {
  String[] inputPaths = new String[] { "normalizeJob",
    "hdfs://192.168.168.101:9000/user/hduser/hdf/MOD13A3.A2005274.h00v10.005.2008079143041.hdf",
    "hdfs://192.168.168.101:9000/user/hduser/test/" };
  ReadMain test = new ReadMain();
  test.runJob(inputPaths);
 }

}

关于MapReduce主程序有几点值得说明一下：

2、本人用的是Yarn2.3.0来执行计算任务，如果用老版本的hadoop，如1.2.0，则把以上主程序中的conf.addResource部分的代码删掉即可。

3、以上MapReduce程序中，只用到了Map函数，未设置Reduce函数。

4、以上程序用到的为HDF4格式的数据，按理说，HDF5格式的数据应该也是支持的。