2.3 HDFS API之旅

本节将提供几个简单的实例，指导读者如何使用Hadoop的Java API针对HDFS进行文件上传、创建、重命名、删除等操作。

1.上传本地文件到HDFS

通过FileSystem.copyFromLocalFile（Path src, Path dst）可将本地文件上传到HDFS的指定位置上，其中src和dst均为文件的完整路径。具体实例如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class CopyFile {
    public static void main(String[] args）throws Exception{
    Configuration conf=new Configuration();
    //conf.addResource(new Path("conf/hadoop-default.xml"));
    //conf.addResource(new Path("conf/hadoop-site.xml"));
    FileSystem hdfs=FileSystem.get(conf);
    Path src=new Path("C:\\Users\\Administrator\\Desktop\\JDK_API_1_6_zh_CN.CHM.1");
    Path dst=new Path("/");
    hdfs.copyFromLocalFile(src, dst);
    System.out.println("Upload to"+conf.get("fs.default.name"));
    FileStatus files[]=hdfs.listStatus(dst);
    for(FileStatus file:files){
       System.out.println(file.getPath());
    }
 }
}

运行结果可以通过控制台和项目浏览器查看，如图2-16和图2-17所示（之后的运行结果将不再展示，读者自行查看）。

2.创建HDFS文件

通过FileSystem.create(Path f)可在HDFS上创建文件，其中f为文件的完整路径。具体实现如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class CreateFile {
    public static void main(String[] args)throws Exception{
    Configuration conf=new Configuration();
    byte[] buff ="hello word!".getBytes();
    FileSystem hdfs = FileSystem.get(conf);
    Path dfs = new Path("/test");
    FSDataOutputStream outputStream = hdfs.create(dfs);
    outputStream.write(buff, 0, buff.length);
    }
}

3.重命名HDFS文件

通过FileSystem.rename(Path src, Path dst)可为指定的HDFS文件重命名，其中src和dst均为文件的完整路径。具体实现如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class Rename {
    public static void main(String[] args)throws Exception{
    Configuration conf=new Configuration();
    FileSystem hdfs=FileSystem.get(conf);
    Path frpath=new Path("/test");
    Path topath=new Path("/test1");
    boolean isRename=hdfs.rename(frpath, topath);
  }
}

4.删除HDFS上的文件

通过FileSystem.delete (Path f, Boolean recursive)可删除指定的HDFS文件，其中f为需要删除文件的完整路径，recursive用来确定是否进行递归删除。具体实现如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class DeleteFile {
public static void main(String[] arg)throws Exception{
      Configuration conf=new Configuration();
      FileSystem hdfs=FileSystem.get(conf);
      Path delef=new Path("/test1");
      boolean isDeleted=hdfs.delete(delef,false);
      //递归删除
      //boolean isDelete=hdfs.delete(delef,true)
      System.out.println("delete?"+isDeleted);
}


}

5.查看HDFS文件的最后修改时间

通过FileStatus.getModificationTime()可查看指定HDFS文件的修改时间。具体实现如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class GetLTime {
    public void main(String[] args)throws Exception{
    Configuration conf=new Configuration();
    FileSystem hdfs=FileSystem.get(conf);
    Path fpath=new Path("/test1");
    FileStatus fileStatus=hdfs.getFileStatus(fpath);
    long modificationTime=fileStatus.getModificationTime();
    System.out.println("Modification time is："+ modificationTime);
}


}

6.查看某个HDFS文件是否存在

通过FileSystem.exists (Path f)可查看指定HDFS文件是否存在，其中f为文件的完整路径。具体实现如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class CheckFile {
public void main(String[] args)throws Exception{
    Configuration conf=new Configuration();
    FileSystem hdfs=FileSystem.get(conf);
    Path findf=new Path("/test1");
    boolean isExists=hdfs.exists(findf);
    System.out.println("Exist?"+isExists);
}


}

7.查找某个文件在HDFS集群的位置

通过FileSystem.getFileBlockLocations(FileStatus file, long start, long len)可查找指定文件在HDFS集群上的位置，其中file为文件的完整路径，start和len来标识查找文件的路径。具体实现如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class FileLoc {
public void main(String[] args)throws Exception{
      Configuration conf=new Configuration();
      FileSystem hdfs=FileSystem.get(conf);
      Path fpath=new Path("/test1");
      FileStatus filestatus=hdfs.getFileStatus(fpath);
      BlockLocation[] blkLocations = hdfs.getFileBlockLocations(filestatus, 0, filestatus.getLen());
      int blockLen=blkLocations.length;
      for(int i=0;i＜blockLen;i++){
      String[]hosts=blkLocations[i].getHosts();
      System.out.println("block"+i+"location："+hosts[i]);
   }


}




}

8.获取HDFS集群上所有节点名称

通过DatanodeInfo.getHostName()可获取HDFS集群上的所有节点名称。具体实现如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
public class GetList {
public void mian(String[] args)throws Exception{
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(conf);
    DistributedFileSystem hdfs = (DistributedFileSystem) fs;
    DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();
    String[] names = new String[dataNodeStats.length];
    for (int i = 0; i < dataNodeStats.length; i++)
    { names[i] = dataNodeStats[i].getHostName();
      System.out.println("node"+i+"name"+ names[i]);
    }
}


}