分布式文件系统HDFS

• 高容错和高可用性，硬件错误是常态而不是异常

HDFS设计运行在普通硬件上，所以硬件故障是很正常的。HDFS提供文件存储副本策略，可以实现错误自检并快速自动恢复。个别硬件的损坏不影响整体数据完整性和计算任务的正常执行。

• 流式数据访问

• 弹性存储，支持大规模数据集

• 简单一致性模型

• 移动计算而非移动数据

• 协议和接口多样性

• 多样的数据管理功能

• 各大运营商

• 中大型互联网公司，如BAT、京东、乐视、美团等

• 金融银行保险类公司

• 各大云平台底层存储平台

• 其他本地系统无法承载存储能力的应用

• HDFS元数据管理者，管理NameSpace(文件系统命名空间)，记录文件是如何分割成数据块以及他们分别存储在集群中的哪些数据节点上。

• NameSpace或其本身属性的任何更改都由NameNode记录，维护整个文件系统的文件和目录。

• DataNode是文件系统的工作节点。根据客户端或者NameNode发送的管理指令，负责HDFS的数据块的读写和检索操作。

• 通过心跳机制定期向NameNode发送他们的存储块的列表。

• 客户端Client代表用户与NameNode或者DataNode交互来访问整个文件系统的对象。

• 开发人员面向Client API来编程实现，对NameNode、DataNode来说透明无感。

• 作用：一是镜像备份，二是日志与镜像的定期备份。两个过程是同时进行的，成为checkpoint。

• 备份镜像的作用：备份fsimage。

• 日志与镜像的定期检查作用：将NameNode中的edits日志与fsimage合并，防止NameNode意外出问题，重启时耗时过长。（主要原因是因为重启时需要重新将fsimage加载进入NN）

• 第一步：Client向NameNode发送数据请求后，寻找数据对应的数据块的位置信息。

• 第二步：NameNode返回文件对应的数据块元数据信息，如所属机器、数据块的block_id、数据块的先后顺序等。

• 第三步：由Client与DataNode直接通信，读取各个block数据块的信息。过程为并行读取，由客户端合并数据。

• 第一步：

• Client向NameNode发送写数据请求后，寻找可以写入的数据块block信息的机器位置。

• 若文件过大，写入可能会分成很多block数据块，实际上是通过一个block一个block的申请。

• 若副本为3，则每次请求后返回一个block的对应的3个副本的block的存放位置。

• 第二步：

• Client获取到对应的block数据块所处的DataNode节点位置后，Client开始写操作。

• Client先写入第一个DataNode，以数据包package的方式逐个发送和接收。如64K大小的package包大小来发送和接收。

• 存在多个副本时，package包的写入是依次进行的。写入到第一个DataNode后，第一个向第二个DataNode传输。第二个写完后，由第二个向第三个DataNode传输package。以此类推。

• 写完一个block数据块后，如果还有则反复进行第一步和第二步。

• 第三步：

• 待所有的数据块block均写完后，Client接收到全部写完的ack答复，告诉NameNode数据已写完，Client关闭socket流。

• DataNode也会向NameNode报告新增block数据块的信息。

hdfs dfs -ls 

hdfs dfs -cat tmp/index.html

hdfs dfs – mkdir tmp/tianliangedu

hdfs dfs -rmr tmp/tianliangedu

hdfs dfs -copyToLocal tmp/index.html .

hdfs dfs -copyFromLocal index.html tmp/tianliangedu

hdfs dfs -text tmp/tianliangedu/index.html.gz | more

hdfs dfs -du -h tmp/tianliangedu

hdfs dfs -touchz tmp/tianliangedu/HelloWorld.txt

hdfs dfs -usage cp

hdfs dfsadmin -help

hdfs dfsadmin -report

hdfs dfsadmin -setQuota <quota> <dirname>……<dirname>

hdfs dfsadmin -safemode get   ##返回安全模式是否开启的信息，返回Safe mode is OFF/OPEN
hdfs dfsadmin -safemode enter ##进入安全模式
hdfs dfsadmin -safemode leave ##强制NameNode退出安全模式
hdfs dfsadmin -safemode wait  ##等待，一直到安全模式结束

hdfs dfs -cat /tmp/jack/input.txt

• 配置依赖jar包的坐标，即描述我是谁。

• 指定依赖的仓库

• 具体需要哪些相关依赖

• 配置打包插件

<project xmlns="http://maven.apache.org/POM/4.0.0";xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";>
   <modelVersion>4.0.0</modelVersion>
   <groupId>com.tianliangedu.course</groupId>
   <artifactId>XjHadoopCore</artifactId>
   <version>0.0.1-SNAPSHOT</version>
   
   <repositories>
        <repository>
           <id>nexus-aliyun</id>
           <name>Nexus aliyun</name>
     <url>http://maven.aliyun.com/nexus/content/groups/public</url>
        </repository>
   </repositories>
   <dependencies>
      
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-client</artifactId>
        <version>2.7.4</version>
        <scope>provided</scope>
      </dependency>
   </dependencies>
   <build>
      <finalName>XjHadoopCore</finalName>
      <plugins>
         <plugin>
           <artifactId>maven-compiler-plugin</artifactId>
           <version>2.3.2</version>
           <configuration>
              <source>1.7</source>
              <target>1.7</target>
              <encoding>UTF-8</encoding>
           </configuration>


        </plugin>
        <plugin>
           <artifactId>maven-assembly-plugin</artifactId>
           <configuration>
              <descriptorRefs>
                 <descriptorRef>jar-with-dependencies</descriptorRef>
              </descriptorRefs>
           </configuration>
           <executions>
              <execution>
                 <id>make-assembly</id>
                 <phase>package</phase>
                 <goals>
                    <goal>assembly</goal>
                 </goals>
              </execution>
           </executions>
        </plugin>
      </plugins>
   </build>
</project>

• 实现步骤

• 定义读取的HDFS文件路径

• 用HDFS Java API将HDFS文本文件转化为字符串形式：

• 通过控制台将字符串打印出来

• 涉及Hadoop中的类说明

• Configuration（重要）

• FileSystem

• Path

• FSDataInputStream

• FSDataOutputStream

• 代码实现

import java.io.ByteArrayOutputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
/**
* hdfs 文件操作工具类,从任意的hdfs filepath中读取文本内容
*/
public class HdfsFileOperatorUtil {
    //添加日志输出能力
    Logger logger = Logger.getLogger(HdfsFileOperatorUtil.class);
    
    // 加载配置文件到内存对象
    static Configuration hadoopConf = new Configuration();
    /** 从HDFS上读取文件 */
    public static String readFromFile(String srcFile) throws Exception {
        //文件路径的空判断
        if (srcFile == null || srcFile.trim().length() == 0) {
            throw new Exception("所要读取的源文件" + srcFile + ",不存在，请检查!");
        }
        //将文件内容转换成字节数组
        byte[] byteArray = readFromFileToByteArray(srcFile);
        if (byteArray == null || byteArray.length == 0) {
            return null;
        }
        //将utf-8编码的字节数组通过utf-8再进行解码
        return new String(byteArray, "utf-8");
    }
    /**
     * 将指定的文件路径从hdfs读取并转换为byte array.
     *
     * @param srcFile
     * @return
     */
    public static byte[] readFromFileToByteArray(String srcFile)
            throws Exception {
        if (srcFile == null || srcFile.trim().length() == 0) {
            throw new Exception("所要读取的源文件" + srcFile + ",不存在，请检查!");
        }
        //获取hadoopConf对应的hdfs集群的对象引用
        FileSystem fs = FileSystem.get(hadoopConf);
        //将给定的srcFile构建成一个hdfs的路径对象Path
        Path hdfsPath=new Path(srcFile);
        FSDataInputStream hdfsInStream = fs.open(hdfsPath);
        
        //初始化一块字节数组缓冲区，大小为65536。缓存每次从流中读取出来的字节数组
        byte[] byteArray = new byte[65536];
        //初始化字节数输出流， 存放最后的所有字节数组
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
            
        // 实际读过来多少
        int readLen = 0;
        //只要还有流数据能读出来，就一直读下去
        while ((readLen = hdfsInStream.read(byteArray)) > 0) {
            bos.write(byteArray);
            byteArray = new byte[65536];
        }
        //读取完成，将hdfs输入流关闭
        hdfsInStream.close();
        //将之前写到字节输出流中的字节，转换成一个整体的字节数组
        byte[] resultByteArray=bos.toByteArray();
       bos.close();
       return resultByteArray;
    }
    
    public static void main(String[] args) throws Exception {
        //定义要读入的hdfs的文件路径
        String hdfsFilePath = "/tmp/jack/input.txt";
        //将文件从hdfs读取下来，转化成字符串
        String result = readFromFile(hdfsFilePath);
        //根据题意，将字符串通过命令行输出
        System.out.println(result);
    }
}

• Maven打包

  右击项目，run as -> maven install进行打包及上传至本地仓库中，生成target目录下的XjHadoopCore-jar-with-dependencies.jar文件。

• 将运行包发布上传至Hadoop环境

• 线上测试运行

yarn jar XjHadoopCore-jar-with-dependencies.jar com.jack.utils.HdfsFileOperatorUtil

• 查看验证效果