# HDFS应用开发规则
#### HDFS NameNode元数据存储路径
NameNode元数据信息的默认存储路径为"${BIGDATA_DATA_HOME}/namenode/data"，该参数用于确定HDFS文件系统的元数据信息的保存路径。
#### HDFS需要开启NameNode镜像备份
NameNode的镜像备份参数为"fs.namenode.image.backup.enable"，需要设置该值为"true"，系统即可定期备份NameNode的数据。
#### HDFS需要开启DataNode数据存储路径
DataNode默认存储路径配置为：${BIGDATA_DATA_HOME}/hadoop/dataN/dn/datadir（N≥1），N为数据存放的目录个数。
例如：${BIGDATA_DATA_HOME}/hadoop/data1/dn/datadir、${BIGDATA_DATA_HOME}/hadoop/data2/dn/datadir
设置后，数据会存储到节点上每个挂载磁盘的对应目录下面。
#### HDFS提高读取写入性能方式
写入数据流程：HDFS Client收到业务数据后，从NameNode获取到数据块编号、位置信息后，联系DataNode，并将需要写入数据的DataNode建立起流水线，完成后，客户端再通过自有协议写入数据到DataNode1，再由DataNode1复制到DataNode2、DataNode3（三备份）。写完的数据，将返回确认信息给HDFS Client。
1. 合理设置块大小，如设置dfs.blocksize为 268435456（即256 MB）。
2. 对于一些不可能重用的大数据，缓存在操作系统的缓存区是无用的。可将以下两参数设置为false： dfs.datanode.drop.cache.behind.reads和dfs.datanode.drop.cache.behind.writes
   
 
#### MapReduce中间文件存放路径
MapReduce默认中间文件夹存放路径只有一个，${hadoop.tmp.dir}/mapred/local，建议修改为每个磁盘下均可存放中间文件。
例如：/hadoop/hdfs/data1/mapred/local、/hadoop/hdfs/data2/mapred/local、/hadoop/hdfs/data3/mapred/local等，不存在的目录会自动忽略。
#### HDFS的大目录操作
不建议对HDFS的大目录进行**du** 、**count** 、**ls** 等操作，如果使用则必须添加-skipLock参数。例如：**hdfs dfs -du -skipLock** *xxx*
#### JAVA开发时，申请资源须在finally释放
申请的HDFS资源需要在try/finally中释放，而不能只在try语句之外释放，否则会导致异常情况下的资源泄漏。
#### HDFS文件操作API概述
Hadoop中关于文件操作类基本上全部是在"org.apache.hadoop.fs"包中，这些API能够支持的操作包含：打开文件，读写文件，删除文件等。Hadoop类库中最终面向用户提供的接口类是FileSystem，该类是个抽象类，只能通过get方法得到具体类。get方法存在几个重载版本，常用的是这个：
```
static FileSystem get(Configuration conf);
```
该类封装了几乎所有的文件操作，例如mkdir，delete等。综上基本可以得出操作文件的程序库框架：
```
operator() 
{
     得到Configuration对象
     得到FileSystem对象
     进行文件操作
 }
```
#### HDFS初始化方法
HDFS初始化是指在使用HDFS提供的API之前，需要做的必要工作。
大致过程为：加载HDFS服务配置文件，并进行Kerberos安全认证，认证通过后再实例化Filesystem，之后使用HDFS的API。此处Kerberos安全认证需要使用到的keytab文件，请提前准备。
正确示例：
```
private  void init() throws IOException {
    Configuration conf = new Configuration();
    // 读取配置文件
    conf.addResource("user-hdfs.xml");
   // 安全模式下，先进行安全认证
    if ("kerberos".equalsIgnoreCase(conf.get("hadoop.security.authentication"))) {
         String PRINCIPAL = "username.client.kerberos.principal";
         String KEYTAB = "username.client.keytab.file";
         // 设置keytab密钥文件
         conf.set(KEYTAB, System.getProperty("user.dir") + File.separator + "conf" + File.separator + conf.get(KEYTAB));
         // 设置kerberos配置文件路径 */
         String krbfilepath = System.getProperty("user.dir") + File.separator + "conf" + File.separator + "krb5.conf";
         System.setProperty("java.security.krb5.conf", krbfilepath);
         // 进行登录认证 */
         SecurityUtil.login(conf, KEYTAB, PRINCIPAL);
     }
     // 实例化文件系统对象
     fSystem = FileSystem.get(conf);
  }
```
#### HDFS上传本地文件
通过FileSystem.copyFromLocalFile (Path src, Path dst)可将本地文件上传到HDFS的指定位置上，其中src和dst均为文件的完整路径。
正确示例：
```
public class CopyFile {
        public static void main(String[] args) throws Exception { 
        Configuration conf=new Configuration();
        FileSystem hdfs=FileSystem.get(conf);
              //本地文件
                 Path src =new Path("D:\\HebutWinOS");
                  //HDFS位置
        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());
              }
        }
 }
```
#### HDFS创建文件
通过FileSystem.mkdirs(Path f)可在HDFS上创建文件夹，其中f为文件夹的完整路径。
正确示例：
```
public class CreateDir { 
    public static void main(String[] args) throws Exception{
         Configuration conf=new Configuration();
         FileSystem hdfs=FileSystem.get(conf);
                Path dfs=new Path("/TestDir");
                hdfs.mkdirs(dfs);
      }
  }
```
#### 查看HDFS文件的最后修改时间
通过FileSystem.getModificationTime()可查看指定HDFS文件的修改时间。
正确示例：
```
public static void main(String[] args) throws Exception {
          Configuration conf=new Configuration();
          FileSystem hdfs=FileSystem.get(conf);
          Path fpath =new Path("/user/hadoop/test/file1.txt");
          FileStatus fileStatus=hdfs.getFileStatus(fpath);
          long modiTime=fileStatus.getModificationTime();
          System.out.println("file1.txt的修改时间是"+modiTime); 
    }
```
#### 读取HDFS某个目录下的所有文件
通过FileStatus.getPath()可查看指定HDFS中某个目录下所有文件。
正确示例：
```
public static void main(String[] args) throws Exception {
         Configuration conf=new Configuration();
         FileSystem hdfs=FileSystem.get(conf);
         Path listf =new Path("/user/hadoop/test");
         FileStatus stats[]=hdfs.listStatus(listf);
         for(int i = 0; i < stats.length; ++i) {
 　　　　　　 System.out.println(stats[i].getPath().toString());
 　　　　}         
         hdfs.close();     
  }
```
#### 查找某个文件在HDFS集群的位置
通过FileSystem.getFileBlockLocation(FileStatus file, long start, long len)可查找指定文件在HDFS集群上的位置，其中file为文件的完整路径，start和len来标识查找文件的路径。
正确示例：
```
public static void main(String[] args) throws Exception {
         Configuration conf=new Configuration();
         FileSystem hdfs=FileSystem.get(conf);
         Path fpath=new Path("/user/hadoop/cygwin");
         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[0]); 
         }
    }
```
#### 获取HDFS集群上所有节点名称信息
通过DatanodeInfo.getHostName()可获取HDFS集群上的所有节点名称。
正确示例：
```
public static void main(String[] args) throws Exception {         
        Configuration conf=new Configuration();         
        FileSystem fs=FileSystem.get(conf);
        
        DistributedFileSystem hdfs = (DistributedFileSystem)fs;         
        DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();                 
        for(int i=0;i < dataNodeStats.length;i++){            
            System.out.println("DataNode_"+i+"_Name:"+dataNodeStats[i].getHostName());         
        }     
   }
```
#### 多线程安全登录方式
如果有多线程进行login的操作，当应用程序第一次登录成功后，所有线程再次登录时应该使用relogin的方式。
login的代码样例：
```
private Boolean login(Configuration conf){     
    boolean flag = false;     
    UserGroupInformation.setConfiguration(conf);          
    try {       
          UserGroupInformation.loginUserFromKeytab(conf.get(PRINCIPAL), conf.get(KEYTAB));       
          System.out.println("UserGroupInformation.isLoginKeytabBased(): " +UserGroupInformation.isLoginKeytabBased());       
          flag = true;     
        } catch (IOException e) {       
           e.printStackTrace();     
        }     
     return flag;        
  }
```
relogin的代码样例：
```
public Boolean relogin(){         
        boolean flag = false;         
        try {                        
              UserGroupInformation.getLoginUser().reloginFromKeytab();           
              System.out.println("UserGroupInformation.isLoginKeytabBased(): " +UserGroupInformation.isLoginKeytabBased());           
              flag = true;         
            } catch (IOException e) {             
                e.printStackTrace();         
            }         
        return flag;     
   }
```
![](https://support.huaweicloud.com/devg-rule-mrs/public_sys-resources/warning_3.0-zh-cn.png)
多次重复登录会导致后建立的会话对象覆盖掉之前登录建立的，将会导致之前建立的会话无法被维护监控，最终导致会话超期后部分功能不可用。