大数据入门第六天——HDFS详解

一、概述

　　1.HDFS中的角色

　　　　Block数据：

　　　　　　HDFS中的文件在物理上是分块存储（block），块的大小可以通过配置参数( dfs.blocksize)来规定，默认大小在hadoop2.x版本中是128M，之前的版本中是64M　

1. 基本存储单位，一般大小为64M（配置大的块主要是因为：1）减少搜寻时间，一般硬盘传输速率比寻道时间要快，大的块可以减少寻道时间；2）减少管理块的数据开销，每个块都需要在NameNode上有对应的记录；3）对数据块进行读写，减少建立网络的连接成本）

2. 一个大文件会被拆分成一个个的块，然后存储于不同的机器。如果一个文件少于Block大小，那么实际占用的空间为其文件的大小

3. 基本的读写S#x5355;位，类似于磁盘的页，每次都是读写一个块

4. 每个块都会被复制到多台机器，默认复制3份

　　　　  NameNode：

　　　　　　负责管理整个文件系统的元数据

　　　　Secondary NameNode：

 　　　　　　定时与NameNode进行同步（定期合并文件系统镜像和编辑日&#x#x5FD7;，然后把合并后的传给NameNode，替换其镜像，并清空编辑日志，类似于CheckPoint机制），但NameNode失效后仍需要手工将其设置成主机——namenode的冷备份

　　　　　　关于这点，可以参考：http://blog.csdn.net/scgaliguodong123_/article/details/46335427

　　　　DataNode：

　　　　　　负责管理用户的文件数据块　　　　

　　　　　文件会按照固定的大小（blocksize）切成若干块（由上传的客户端进行切块处理，这样不大于128M切块大小的实际是多少就是多少）后分布式存储在若干台datanode上

　　　　　　Datanode会定期向Namenode汇报自身所保存的文件block信息，而namenode则会负责保持文件的副本数量

　　详细角色信息，参考：https://www.w3cschool.cn/hadoop/xvmi1hd6.html

　　漫画式的讲解，参考：https://www.cnblogs.com/raphael5200/p/5497218.html

二、读写数据流程

　　简要的说明参考上文漫画式讲解处

　　专业深入讲解，参考：https://www.cnblogs.com/codeOfLife/p/5375120.html

　　1.写数据流程

1、根namenode通信请求上传文件，namenode检查目标文件是否已存在，父目录是否存在
2、namenode返回是否可以上传
3、client请求第一个 block该传输到哪些datanode服务器上
4、namenode返回3个datanode服务器ABC
5、client请求3台dn中的一台A上传数据（本质上是一个RPC调用，建立pipeline），A收到请求会继续调用B，然后B调用C，将真个pipeline建立完成，逐级返回客户端
6、client开始往A上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位，A收到一个packet就会传给B，B传给C；A每传一个packet会放入一个应答队列等待应答
7、当一个block传输完成之后，client再次请求namenode上传第二个block的服务器。

　　图解请参考上文

　　注意：HDFS只允许修改文件名/文件追加等，无法直接修改原来的文件！

　　2.读数据流程

1、跟namenode通信查询元数据，找到文件块所在的datanode服务器
2、挑选一台datanode（就近原则，然后随机）服务器，请求建立socket流
3、datanode开始发送数据（从磁盘里面读取数据放入流，以packet为单位来做校验）
4、客户端以packet为单位接收，现在本地缓存，然后写入目标文件

 三、HDFS元数据管理与解析

　　1.元数据管理原理

　　　　元数据分类：

• 第一类是文件和目录自身的属性信息，例如文件名、目录名、父目录信息、文件大小、创建时间、修改时间等。
• 第二类记录文件内容存储相关信息，例如文件块情况、副本个数、每个副本所在的Data Node 信息等。
• 第三类用来记录HDFS中所有Data Node信息，用于Data Node管理。

　　　　存储机制：

　　　　　　A、内存中有一份完整的元数据(内存meta data)

　　　　　　B、磁盘有一个“准完整”的元数据镜像（fsimage）文件(在namenode的工作目录中)，整个运行过程中,fsimage是不会改变的！editslog的更新会同步到内存

　　　　　　C、用于衔接内存metadata和持久化元数据镜像fsimage之间的操作日志（edits文件）注：当客户端对hdfs中的文件进行新增或者修改操作，操作记录首先被记入edits日志文件中，当客户端操作成功后，相应的元数据会更新到内存meta.data中

　　　　更多参考：http://blog.csdn.net/xiaming564/article/details/23165253

　　　　　　　　http://blog.csdn.net/chenkfkevin/article/details/61196409

 　　　　这里就能很清楚的知道secondary namenode的工作原理了！

 　　　　关于元数据的存取流程与分析，参考网友的白话讲解：http://blog.csdn.net/lepton126/article/details/53183037

 　　　　2.修改工作目录

　　　　之前我们配置过hadoop.tmp.dir来设置临时目录，这里可以通过以下参数设置HDFS工作目录

配置文件：hdfs-site.xml
参数名：dfs.namenode.name.dir    
格式：file://${hadoop.tmp.dir}/dfs/name    
说明：Determines where on the local filesystem the DFS name node should store the name table(fsimage). 
If this is a comma-delimited list of directories then the name table is replicated in all of the directories, for redundancy.
示例：
    <property>
        <name>dfs.namenode.name.dir</name>
        <value>/home/hadoop/name1,/home/hadoop/name2</value>
    </property>    

　　//当然，dfs.datanode.data.dir也是可以配置的，这样重新格式化后就可以重新使用工作目录了！点击查看对比

　　注：如果还在使用dfs.name.dir/dfs.data.dir，请查看官网配置的deprecated properties

 　　　　3.元数据各目录解析

　　　　　　详细解析，参考：https://www.iteblog.com/archives/967.html

 　　　　　　　　　　　　　　http://blog.csdn.net/opensure/article/details/51452058

四、HDFS的Java-API操作

　　 完整API参考官网：http://hadoop.apache.org/docs/current/api/

　　1.基本的增删改查API：

package com.hdfs.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;

import java.net.URI;

/**
 * HDFS客户端demo
 *
 * @author zcc ON 2018/1/28
 **/
public class HdfsClientDemo {
    FileSystem fs = null;
    @Before
    public void init() throws Exception{
        // 如配置完环境变量，下行可省略
        // System.setProperty("hadoop.home.dir", "F:\work\hadoop-2.6.4");
        Configuration conf = new Configuration();
        // 配置文件系统（注意hosts的配置）
        // conf.set("fs.defaultFS","hdfs://mini1:9000");
        // 拿到一个操作的客户端实例对象（此处使用3个参数则上一行省略）
        fs = FileSystem.get(new URI("hdfs://mini1:9000"),conf,"hadoop");
    }
    @Test
    public void testUpload() throws Exception{
        // 就对应get的别名
        fs.copyFromLocalFile(new Path("F:/c.log"),new Path("/c.log.copy"));
        // 关闭
        fs.close();
    }
    @Test
    public void testDelete() throws Exception{
        // 第一个是Path，第二个为是否递归删除
        boolean b = fs.delete(new Path("/c.log.copy"), true);
        System.out.println("删除状态：" + b);
        fs.close();
    }
    @Test
    public void testList() throws Exception{
        /*
        一般而言，大数据方面使用迭代器场景居多，因为Iterator它本身并不存数据（可以查看源码）
        它只是提供了几个简单的方法帮你去取数据，而使用ArrayList则是直接把数据拿过来了，大数据
        量下不适合
         */
        RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path("/"), true);
        while (iterator.hasNext()) {
            LocatedFileStatus next =  iterator.next();
            System.out.println("Path:" + next.getPath());
            System.out.println("BlockSize" + next.getBlockSize());
            System.out.println("Name:" + next.getPath().getName());
        }

//        Path:hdfs://mini1:9000/1.txt
//        BlockSize134217728
//        Name:1.txt
        fs.close();
    }

    /**
     * 既可以遍历文件，又可以遍历文件夹
     * @throws Exception
     */
    @Test
    public void testList2() throws Exception{
        FileStatus[] files = fs.listStatus(new Path("/"));
        for (FileStatus file : files) {
            System.out.println(file.getPath().getName());
            if (file.isFile()) {
                System.out.println("it is a file");
            }
        }
    }
}

 　　这里引入Java中迭代器的机制浅谈，供参考：https://www.cnblogs.com/hasse/p/5024193.html

　　更多实例，参考：http://blog.csdn.net/litianxiang_kaola/article/details/70983904

　　2.流操作API

　　　　相对那些封装好的方法而言的更底层一些的操作方式上层那些mapreduce   spark等运算框架，去hdfs中获取数据的时候，就是调的这种底层的api

package com.hdfs.stream;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;

/**
 * HDFS的流API操作
 *
 * @author zcc ON 2018/1/30
 **/
public class HdfsStreamAccess {
    FileSystem fs = null;
    @Before
    public void init() throws Exception{
        Configuration conf = new Configuration();
        fs = FileSystem.get(new URI("hdfs://mini1:9000"),conf,"hadoop");
    }

    /**
     * 通过流的操作，就可以为上层MR程序提供服务，例如跑任务只需要跑60M，这样就不用通过
     * 之前的直接得到一整个文件的复杂处理
     * 而建立文件夹等直接通过简单API即可！
     * @throws Exception
     */
    @Test
    public void testUpload() throws Exception{
        // 输出到HDFS
        FSDataOutputStream outputStream = fs.create(new Path("/angle"), true);
        // 本地输入流
        FileInputStream inputStream = new FileInputStream("F:\c.log");
        // 通过IOUtils进行拷贝（使用更加通用的IOUtils）
        IOUtils.copy(inputStream, outputStream);
    }
    @Test
    public void testDownload() throws Exception{
        FSDataInputStream inputStream = fs.open(new Path("/angle"));
        FileOutputStream outputStream = new FileOutputStream("F:\c-download.log");
        IOUtils.copy(inputStream, outputStream);
    }

    /**
     * 指定随机长度读取
     */
    @Test
    public void testRandomAccess() throws Exception{
        FSDataInputStream inputStream = fs.open(new Path("/angle"));
        inputStream.seek(12);
        FileOutputStream outputStream = new FileOutputStream("F:\c-random.log");
        // 从12字节读到末尾（也可以通过while结合自定义的count等来控制读文件的大小）
        // 后续会避免读取读到单词一半这样的问题
//        IOUtils.copy(inputStream, outputStream);
        IOUtils.copyLarge(inputStream, outputStream, 12, 100);
    }

    /**
     * 以下模拟实现：获取一个文件的所有block位置信息，然后读取指定block中的内容
     * @throws IllegalArgumentException
     * @throws IOException
     */
    @Test
    public void testCat() throws IllegalArgumentException, IOException {

        FSDataInputStream in = fs.open(new Path("/weblog/input/access.log.10"));
        //拿到文件信息
        FileStatus[] listStatus = fs.listStatus(new Path("/weblog/input/access.log.10"));
        //获取这个文件的所有block的信息
        BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(listStatus[0], 0L, listStatus[0].getLen());
        //第一个block的长度
        long length = fileBlockLocations[0].getLength();
        //第一个block的起始偏移量
        long offset = fileBlockLocations[0].getOffset();

        System.out.println(length);
        System.out.println(offset);

        //获取第一个block写入输出流
//        IOUtils.copyBytes(in, System.out, (int)length);
        byte[] b = new byte[4096];

        FileOutputStream os = new FileOutputStream(new File("d:/block0"));
        while(in.read(offset, b, 0, 4096)!=-1){
            os.write(b);
            offset += 4096;
            if(offset>=length) return;
        };
        os.flush();
        os.close();
        in.close();
    }
}

 五、案例：shell脚本日志采集

　　采用shell编写的脚本如下：

#!/bin/bash

#set java env
export JAVA_HOME=/home/hadoop/app/jdk1.7.0_51
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

#set hadoop env
export HADOOP_HOME=/home/hadoop/app/hadoop-2.6.4
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH


#版本1的问题：
#虽然上传到Hadoop集群上了，但是原始文件还在。如何处理？
#日志文件的名称都是xxxx.log1,再次上传文件时，因为hdfs上已经存在了，会报错。如何处理？

#如何解决版本1的问题
#       1、先将需要上传的文件移动到待上传目录
#    2、在讲文件移动到待上传目录时，将文件按照一定的格式重名名
#        /export/software/hadoop.log1   /export/data/click_log/xxxxx_click_log_{date}


#日志文件存放的目录
log_src_dir=/home/hadoop/logs/log/

#待上传文件存放的目录
log_toupload_dir=/home/hadoop/logs/toupload/


#日志文件上传到hdfs的根路径
hdfs_root_dir=/data/clickLog/20151226/

#打印环境变量信息
echo "envs: hadoop_home: $HADOOP_HOME"


#读取日志文件的目录，判断是否有需要上传的文件
echo "log_src_dir:"$log_src_dir
ls $log_src_dir | while read fileName
do
    if [[ "$fileName" == access.log.* ]]; then
    # if [ "access.log" = "$fileName" ];then
        date=`date +%Y_%m_%d_%H_%M_%S`
        #将文件移动到待上传目录并重命名
        #打印信息
        echo "moving $log_src_dir$fileName to $log_toupload_dir"xxxxx_click_log_$fileName"$date"
        mv $log_src_dir$fileName $log_toupload_dir"xxxxx_click_log_$fileName"$date
        #将待上传的文件path写入一个列表文件willDoing
        echo $log_toupload_dir"xxxxx_click_log_$fileName"$date >> $log_toupload_dir"willDoing."$date
    fi
    
done
#找到列表文件willDoing
ls $log_toupload_dir | grep will |grep -v "_COPY_" | grep -v "_DONE_" | while read line
do
    #打印信息
    echo "toupload is in file:"$line
    #将待上传文件列表willDoing改名为willDoing_COPY_
    mv $log_toupload_dir$line $log_toupload_dir$line"_COPY_"
    #读列表文件willDoing_COPY_的内容（一个一个的待上传文件名）  ,此处的line 就是列表中的一个待上传文件的path
    cat $log_toupload_dir$line"_COPY_" |while read line
    do
        #打印信息
        echo "puting...$line to hdfs path.....$hdfs_root_dir"
        hadoop fs -put $line $hdfs_root_dir
    done    
    mv $log_toupload_dir$line"_COPY_"  $log_toupload_dir$line"_DONE_"
done

　　加入定时任务调度，参考大数据之Linux基础