Java 操作HDFS小结

/**
 * create a existing file from local filesystem to hdfs
 * @param source
 * @param dest
 * @param conf
 * @throws IOException
 */
public void addFile(String source, String dest, Configuration conf) throws IOException {

  FileSystem fileSystem = FileSystem.get(conf);
  // Get the filename out of the file path
  String filename = source.substring(source.lastIndexOf('/') + 1,source.length());
  // Create the destination path including the filename.
  if (dest.charAt(dest.length() - 1) != '/') {
    dest = dest + "/" + filename;
  } else {
    dest = dest + filename;
  }
  // Check if the file already exists
  Path path = new Path(dest);
  if (fileSystem.exists(path)) {
    System.out.println("File " + dest + " already exists");
    return;
  }
  // Create a new file and write data to it.
  FSDataOutputStream out = fileSystem.create(path);
  InputStream in = new BufferedInputStream(new FileInputStream(new File(
      source)));
  byte[] b = new byte[1024];
  int numBytes = 0;
  while ((numBytes = in.read(b)) > 0) {
    out.write(b, 0, numBytes);
  }
  // Close all the file descriptors
  in.close();
  out.close();
}

/**
 * read a file from hdfs
 * @param file
 * @param conf
 * @throws IOException
 */
public void readFile(String file, Configuration conf) throws IOException {
  FileSystem fileSystem = FileSystem.get(conf);
  Path path = new Path(file);
  if (!fileSystem.exists(path)) {
    System.out.println("File " + file + " does not exists");
    return;
  }
  FSDataInputStream in = fileSystem.open(path);

  String filename = file.substring(file.lastIndexOf('/') + 1,
      file.length());

  OutputStream out = new BufferedOutputStream(new FileOutputStream(
      new File(filename)));

  byte[] b = new byte[1024];
  int numBytes = 0;
  while ((numBytes = in.read(b)) > 0) {
    out.write(b, 0, numBytes);
  }
  in.close();
  out.close();
}

/**
 * delete a directory in hdfs
 * @param file
 * @throws IOException
 */
public void deleteFile(String file, Configuration conf) throws IOException {
  FileSystem fileSystem = FileSystem.get(conf);

  Path path = new Path(file);
  if (!fileSystem.exists(path)) {
    System.out.println("File " + file + " does not exists");
    return;
  }
  fileSystem.delete(new Path(file), true);
}

/**
 * create directory in hdfs
 * @param dir
 * @throws IOException
 */
public void mkdir(String dir, Configuration conf) throws IOException {
  FileSystem fileSystem = FileSystem.get(conf);
  Path path = new Path(dir);
  if (fileSystem.exists(path)) {
    System.out.println("Dir " + dir + " already not exists");
    return;
  }
  fileSystem.mkdirs(path);
}
/**
 * list file status
 *
 * @param path file path
 * @return file status
 */
public List<FileStatus> listFile(String path) throws IOException {
  List<FileStatus> files = new ArrayList<FileStatus>();
  FileSystem fileSystem = FileSystem.get(conf);
  Path fsPath = new Path(path);
  FileStatus[] sts;
  try {
    FileStatus file = fileSystem.getFileStatus(fsPath);
    if (file.isDirectory()) {
      sts = fileSystem.listStatus(fsPath);
    } else {
      sts = new FileStatus[] {file};
    }
  } catch (Exception ex) {
    // may be Wildcard
    sts = getFileSystem().globStatus(fsPath);
  }
  if (sts != null) {
    Collections.addAll(files, sts);
  }
  return files;
}
/**
 * get file size
 *
 * @param fold file path
 * @return file size
 */
public long fileSize(String fold) {
  long len = 0L;
  List<FileStatus> list = listFile(fold.trim());
  for (FileStatus fileStatus : list) {
    len += fileStatus.getLen();
  }
  return len;
}

public class FileSystemFactory implements PooledObjectFactory<FileSystem> {

    private final Configuration conf;

    public FileSystemFactory(Configuration conf){
        this.conf = conf;
    }

    @Override
    public PooledObject<FileSystem> makeObject() throws Exception {
        return new DefaultPooledObject<FileSystem>(FileSystem.get(conf));
    }

    @Override
    public void destroyObject(PooledObject<FileSystem> pooledObject) throws Exception {
        FileSystem hdfs = pooledObject.getObject();
        hdfs.close();
    }

    @Override
    public boolean validateObject(PooledObject<FileSystem> pooledObject) {
        FileSystem hdfs = pooledObject.getObject();
        try {
            return  hdfs.exists(new Path("/"));
        } catch (IOException e) {
            return false;
        }
    }
    @Override
    public void activateObject(PooledObject<FileSystem> pooledObject) throws Exception {}

    @Override
    public void passivateObject(PooledObject<FileSystem> pooledObject) throws Exception {}}
}
public class HdfsPoolConfig extends GenericObjectPoolConfig<FileSystem> {
  public HdfsPoolConfig HdfsPoolConfig(
    Integer maxTotal,Integer maxIdle,Integer minIdle,
    Long maxWaitMillis,Long minEvictableIdleTimeMillis,
    Long timeBetweenEvictionRunsMillis
  ){
    HdfsPoolConfig hdfsPoolConfig = new HdfsPoolConfig();
    hdfsPoolConfig.setMaxTotal(maxTotal);
    hdfsPoolConfig.setMaxIdle(maxIdle);
    hdfsPoolConfig.setMinIdle(minIdle));
    hdfsPoolConfig.setMaxWaitMillis(maxWaitMillis);
    hdfsPoolConfig.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
    hdfsPoolConfig.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);
    return hdfsPoolConfig;
  }
}
public class HdfsPool extends GenericObjectPool<FileSystem> {
    public HdfsPool(PooledObjectFactory<FileSystem> factory) {
        super(factory);
    }
    public HdfsPool(PooledObjectFactory<FileSystem> factory, GenericObjectPoolConfig<FileSystem> config) {
        super(factory, config);
    }
    public HdfsPool(PooledObjectFactory<FileSystem> factory, GenericObjectPoolConfig<FileSystem> config, AbandonedConfig abandonedConfig) {
        super(factory, config, abandonedConfig);
    }
}