Hadoop基础知识：分布式文件系统_分布式文件系统hdfs是什么,什么场景下使用比较合适?

分布式文件系统，英文全称为Hadoop Distribute File System；简称：HDFS。是hadoop核心组件之一，作为最底层的分布式存储服务而存在。分布式文件系统解决的问题就是大数据存储。它们是横跨在多台计算机上的存储系统。分布式文件系统在大数据时代有着广泛的应用前景，它们为存储和处理超大规模数据提供所需的扩展能力。

一、应用场景

1 适合场景

存储超大文件： 这里超大指的是几百MB、GB、或者TB级别，需要高吞吐量，对延时没有要求；

采用流式数据访问方式： 即一次写入、多次读取。数据集经常从数据源生成或者拷贝一次，然后在其上做很多分析工作 ；

运行于商业硬件上： Hadoop不需要特别贵的机器，可运行于普通廉价机器，可以节约成本；也因此具有比较高的故障率，需要高容错性；

为数据存储提供所需的扩展能力；

2 不适合场景

低延时的数据访问： 对延时要求在毫秒级别的应用，不适合采用HDFS。HDFS是为高吞吐数据传输设计的,因此会以提高延时为代价；

大量小文件： 文件的元数据保存在NameNode的内存中， 整个文件系统的文件数量会受限于NameNode的内存大小。 经验而言，一个文件/目录/文件块一般占有150字节的元数据内存空间。如果有100万个文件，每个文件占用1个文件块，则需要大约300M的内存。因此十亿级别的文件数量在现有商用机器上难以支持；

多方读写，需要任意的文件修改： HDFS采用追加（append-only）的方式写入数据，不支持文件任意位置的修改，不支持多个写入器；

二、核心概念

三、命令行接口

1 常用命令

HDFS提供了各种交互方式，无疑命令行是最简单的，下面总结一些常用的交互命令。在bash窗口输入hdfs dfs [generic options](反正就是一个未知命令就行)，会显示出HDFS命令行使用说明。

ls： 查看目录下内容，包括文件名，权限，所有者，大小和修改时间

$ hdfs dfs -ls [-R] <args>
1

mkdir：创建目录

$ hdfs dfs -mkdir [-p] <paths>
1

moveFromLocal：将文件从本地剪切到hdfs

$ hdfs dfs -moveFromLocal <localsrc> <dest>
1

mv：将文件从源移动到目标；此命令还允许多个源，在这种情况下，目标需要是一个目录。不允许跨文件系统移动文件。

$ hdfs dfs -mv URI [URI...] <dest>
1

put：将本地文件系统中的一个或多个目录复制到目标文件系统，还可以从stdin读取输入并将其写入目标文件系统。

$ hdfs dfs -put <localsrc> ... <dst>
1

appendToFile：追加一个或者多个文件到hdfs指定文件中，也可以从命令行读取输入。

$ hdfs dfs -appendToFile <localsrc> ... <dest>
1

cat：查看内容

$ hdfs dfs -cat URI [URI ...]
1

cp：复制文件(夹)，可以覆盖，可以保留原有权限信息

$ hdfs dfs -cp [-f] [-p | -p[topax]] URI [URI ...] <dest>
1

rm：删除指定参数得文件

$ hdfs dfs -rm [-f] [-r | -R] [-skipTrash] URI [URI ...]
1

chmod：修改权限

$ hdfs dfs -chmod [-R] <MODE[,MODE]... | OCTALMODE> URI [URI ...]
1

chown：修改所有者

$ hdfs dfs -chmod [-R] [OWNER][:[GROUP]] URI [URI]
1

expunge：清空回收站

$ hdfs dfs -expunge
1

2 文件限额配置命令

hdfs文件的限额配置允许我们以文件大小或者文件个数来限制我们在某个目录下上传的文件数量或者文件内容总量，以便达到我们类似百度网盘等限制每个用户允许上传的最大的文件的量

数量限额

$ hdfs dfs -mkdir -p /user/root/list
$ hdfs dfsadmin -setQuota 2 list  ## 给该文件夹下面设置最多上传两个文件
$ hdfs dfsadmin -clrQuota /user/root/list ## 清除文件数量限制
1
2
3

空间大小限额

$ hdfs dfsadmin -setSpaceQuota 4k /user/root/list ## 限制空间大小4kb
$ hdfs dfsadmin -clrSpaceQuota /user/root/list ## 清除空间限额
1
2

查看文件限额数量

$ hdfs dfs -count -q -h /user/root/list
## 该查询命令显示八个参数，分别是
## 前四个： 文件个数配额，文件个数配额余量，空间配置，空间剩余配置
## 后四个： 文件夹个数，文件个数，内容大小，路径
1
2
3
4

图中表示，文件配额个数为3个，剩余0个；空间为4MB，剩余4MB；文件夹1个；已上传文件个数为2个；上传文件的大小为6.7kb; 限额的路径为/user/root/list

3 安全模式

安全模式是HDFS所处的一种特殊状态，在这种状态下，文件系统只接受读数据请求，而不接受删除、修改等变更请求。在NameNode主节点启动时，HDFS首先进入安全模式，DataNode在启动的时候会向namenode汇报可用的block等状态，当整个系统达到安全标准时，HDFS自动离开安全模式。如果HDFS出于安全模式下，则文件block不能进行任何的副本复制操作，因此达到最小的副本数量要求是基于datanode启动时的状态来判定的，启动时不会再做任何复制（从而达到最小副本数量要求），hdfs集群刚启动的时候，默认30S钟的时间是出于安全期的，只有过了30S之后，集群脱离了安全期，然后才可以对集群进行操作。

$ hdfs dfsadmin [-safemode enter | leave | get | wait] 
1

四、Hadoop文件系统

Hadoop有一个抽象的文件系统概念，HDFS只是其中的一个实现。java抽象类org.apache.hadoop.fs.FileSystem定义了Hadoop中一个文件系统的客户端接口，该抽象类有以下几个具体的实现：

五、HDFS的API操作

maven需要导入的依赖包

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>2.7.5</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.7.5</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>2.7.5</version>
</dependency>

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>2.7.5</version>
</dependency>
<!-- https://mvnrepository.com/artifact/junit/junit -->
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.11</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testng</groupId>
    <artifactId>testng</artifactId>
    <version>RELEASE</version>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

添加如下maven插件：

<plugins>
    <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.0</version>
        <configuration>
            <source>1.8</source>
            <target>1.8</target>
            <encoding>UTF-8</encoding>
            <!-- <verbal>true</verbal>-->
        </configuration>
    </plugin>
    <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>3.2.1</version>
        <executions>
            <execution>
                <phase>package</phase>
                <goals>
                    <goal>shade</goal>
                </goals>
                <configuration>
                    <minimizeJar>true</minimizeJar>
                </configuration>
            </execution>
        </executions>
    </plugin>
</plugins>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

如果使用的是CDH集群，由于cdh版本的所有的软件涉及版权的问题，所以并没有将所有的jar包托管到maven仓库当中去，而是托管在了CDH自己的服务器上面，所以我们默认去maven的仓库下载不到，需要自己手动的添加repository去CDH仓库进行下载，以下两个地址是官方文档说明，请仔细查阅

https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo.html

在pom.xml文件中添加CDH仓库

<repositories>
    <repository>
        <id>cloudera</id>
        <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
    </repository>
</repositories>
1
2
3
4
5
6

1 使用URI方式访问数据

@Test
public void demo1()throws  Exception{
    //第一步：注册hdfs 的url，让java代码能够识别hdfs的url形式
    URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());

    InputStream inputStream = null;
    FileOutputStream outputStream =null;
    //定义文件访问的url地址
    String url = "hdfs://192.168.17.100:8020/test/input/install.log";
    //打开文件输入流
    try {
        inputStream = new URL(url).openStream();
        outputStream = new FileOutputStream(new File("c:\\hello.txt"));
        IOUtils.copy(inputStream, outputStream);
    } catch (IOException e) {
        e.printStackTrace();
    }finally {
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

2 使用文件系统方式访问数据

在 java 中操作 HDFS，主要涉及以下 Class：

• Configuration：该类的对象封装了客户端或者服务器的配置;

• FileSystem：该类的对象是一个文件系统对象，可以用该对象的一些方法来对文件进行操作，通过 FileSystem 的静态方法 get 获得该对象。

  FileSystem fs = FileSystem.get(conf)

get 方法从conf中的一个参数fs.defaultFS的配置值判断具体是什么类型的文件系统。如果我们的代码中没有指定fs.defaultFS，并且工程classpath下也没有给定相应的配置，conf中的默认值就来自于hadoop的jar包中的core-default.xml，默认值为：file:///，则获取的将不是一个DistributedFileSystem的实例，而是一个本地文件系统的客户端对象。

2.1 获取FileSystem的几种方式

第一种方式：

@Test
public void getFileSystem() throws URISyntaxException, IOException {
   Configuration configuration = new Configuration();
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), configuration);
    System.out.println(fileSystem.toString());
}
1
2
3
4
5
6

第二种方式：

@Test
public void getFileSystem2() throws URISyntaxException, IOException {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS","hdfs://192.168.17.100:8020");
    FileSystem fileSystem = FileSystem.get(new URI("/"), configuration);
    System.out.println(fileSystem.toString());
}
1
2
3
4
5
6
7

第三种方式：

@Test
public void getFileSystem3() throws URISyntaxException, IOException {
    Configuration configuration = new Configuration();
    FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://192.168.17.100:8020"), configuration);
    System.out.println(fileSystem.toString());
}
1
2
3
4
5
6

第四种方式：

@Test
public void getFileSystem4() throws  Exception{
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS","hdfs://192.168.17.100:8020");
    FileSystem fileSystem = FileSystem.newInstance(configuration);
    System.out.println(fileSystem.toString());
}
1
2
3
4
5
6
7

2.2 递归遍历文件系统中的所有文件

通过递归遍历hdfs文件系统

@Test
public void listFile() throws Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), new Configuration());
    FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
    for (FileStatus fileStatus : fileStatuses) {
        if(fileStatus.isDirectory()){
            Path path = fileStatus.getPath();
            listAllFiles(fileSystem,path);
        }else{
            System.out.println("文件路径为"+fileStatus.getPath().toString());

        }
    }
}
public void listAllFiles(FileSystem fileSystem,Path path) throws  Exception{
    FileStatus[] fileStatuses = fileSystem.listStatus(path);
    for (FileStatus fileStatus : fileStatuses) {
        if(fileStatus.isDirectory()){
            listAllFiles(fileSystem,fileStatus.getPath());
        }else{
            Path path1 = fileStatus.getPath();
            System.out.println("文件路径为"+path1);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

官方提供API直接遍历

@Test
public void listMyFiles()throws Exception{
    //获取fileSystem类
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), new Configuration());
    //获取RemoteIterator 得到所有的文件或者文件夹，第一个参数指定遍历的路径，第二个参数表示是否要递归遍历
    RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);
    while (locatedFileStatusRemoteIterator.hasNext()){
        LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
        System.out.println(next.getPath().toString());
    }
    fileSystem.close();
}
1
2
3
4
5
6
7
8
9
10
11
12

2.3 下载文件到本地

@Test
public void getFileToLocal()throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), new Configuration());
    FSDataInputStream open = fileSystem.open(new Path("/test/input/install.log"));
    FileOutputStream fileOutputStream = new FileOutputStream(new File("c:\\install.log"));
    IOUtils.copy(open,fileOutputStream );
    IOUtils.closeQuietly(open);
    IOUtils.closeQuietly(fileOutputStream);
    fileSystem.close();
}
1
2
3
4
5
6
7
8
9
10

2.4 hdfs上创建文件夹

@Test
public void mkdirs() throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), new Configuration());
    boolean mkdirs = fileSystem.mkdirs(new Path("/hello/mydir/test"));
    fileSystem.close();
}
1
2
3
4
5
6

2.5 hdfs文件上传

@Test
public void putData() throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), new Configuration());
    fileSystem.copyFromLocalFile(new Path("file:///c:\\install.log"),new Path("/hello/mydir/test"));
    fileSystem.close();
}
1
2
3
4
5
6

3 hdfs权限问题以及伪造用户

首先停止hdfs集群，在node01机器上执行以下命令

$ cd /usr/local/hadoop-2.6.0-cdh5.14.0
$ sbin/stop-dfs.sh
1
2

修改node01机器上的hdfs-site.xml当中的配置文件

$ cd /usr/local/hadoop-2.6.0-cdh5.14.0/etc/hadoop
$ vim hdfs-site.xml
1
2

<property>
  <name>dfs.permissions</name>
  <value>true</value>
</property>
1
2
3
4

修改完成之后配置文件发送到其他机器上面去

$ scp hdfs-site.xml node02:$PWD
$ scp hdfs-site.xml node03:$PWD
1
2

重启hdfs集群

$ cd /.../hadoop-2.7.5
$ sbin/start-dfs.sh
1
2

随意上传一些文件到我们hadoop集群当中准备测试使用

$ cd /export/servers/hadoop-2.7.5/etc/hadoop
$ hdfs dfs -mkdir /config
$ hdfs dfs -put *.xml /config
$ hdfs dfs -chmod 600 /config/core-site.xml  
1
2
3
4

使用代码准备下载文件

@Test
public void getConfig()throws Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), new Configuration(),"root");
    fileSystem.copyToLocalFile(new Path("/config/core-site.xml"),new Path("file:///c:/core-site.xml"));
    fileSystem.close();
}
1
2
3
4
5
6

4 hdfs的小文件合并

由于hadoop擅长存储大文件，因为大文件的元数据信息比较少，如果hadoop集群当中有大量的小文件，那么每个小文件都需要维护一份元数据信息，会大大的增加集群管理元数据的内存压力，所以在实际工作当中，如果有必要一定要将小文件合并成大文件进行一起处理。在hdfs 的shell命令模式下，可以通过命令行将很多的hdfs文件合并成一个大文件下载到本地，命令如下：

$ cd /usr/local/
$ hdfs dfs -getmerge /config/*.xml ./hello.xml
1
2

既然可以在下载的时候将这些小文件合并成一个大文件一起下载，那么肯定就可以在上传的时候将小文件合并到一个大文件里面去，代码如下：

//将多个本地系统文件，上传到hdfs，并合并成一个大的文件
@Test
public void mergeFile() throws  Exception{
    //获取分布式文件系统
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.17.100:8020"), new Configuration(),"root");
    FSDataOutputStream outputStream = fileSystem.create(new Path("/bigfile.xml"));
    //获取本地文件系统
    LocalFileSystem local = FileSystem.getLocal(new Configuration());
    //通过本地文件系统获取文件列表，为一个集合
    FileStatus[] fileStatuses = local.listStatus(new Path("file:///F:\\上传小文件合并"));
    for (FileStatus fileStatus : fileStatuses) {
        FSDataInputStream inputStream = local.open(fileStatus.getPath());
       IOUtils.copy(inputStream,outputStream);
        IOUtils.closeQuietly(inputStream);
    }
    IOUtils.closeQuietly(outputStream);
    local.close();
    fileSystem.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

六、数据流

1 HDFS的文件副本机制及block块存储

所有的文件都是以block块的方式存放在HDFS文件系统当中，在hadoop1当中，文件的block块默认大小是64M，hadoop2当中，文件的block块大小默认是128M，block块的大小可以通过hdfs-site.xml当中的配置文件进行指定

<property>
  <name>dfs.block.size</name>
  <value>块大小 以字节为单位</value>//只写数值就可以
</property>
1
2
3
4

抽象成数据块的好处

1. 一个文件有可能大于集群中任意一个磁盘
2. 使用块抽象而不是文件可以简化存储子系统
3. 块适合用于数据备份进而提供数据容错能力和可用性

块缓存

通常DataNode从磁盘中读取块，但对于访问频繁的文件，其对应的块可能被显示的缓存在DataNode的内存中，以堆外块缓存的形式存在。默认情况下，一个块仅缓存在一个DataNode的内存中，当然可以针对每个文件配置DataNode的数量。作业调度器通过在缓存块的DataNode上运行任务，可以利用块缓存的优势提高读操作的性能。

HDFS的文件权限验证

hdfs的文件权限机制与linux系统的文件权限机制类似：r:read w:write x:execute 权限x对于文件表示忽略，对于文件夹表示是否有权限访问其内容

如果linux系统用户zhangsan使用hadoop命令创建一个文件，那么这个文件在HDFS当中的owner就是zhangsan。

HDFS文件权限的目的，防止好人做错事，而不是阻止坏人做坏事。HDFS相信你告诉我你是谁，你就是谁

2 HDFS的文件写入过程

详细步骤解析：

1. client发起文件上传请求，通过RPC与NameNode建立通讯，NameNode检查目标文件是否已存在，父目录是否存在，返回是否可以上传；

2. client请求第一个block该传输到哪些DataNode服务器上；

3. NameNode根据配置文件中指定的备份数量及机架感知原理进行文件分配，返回可用的DataNode的地址如：A，B，C；

   注：Hadoop在设计时考虑到数据的安全与高效，数据文件默认在HDFS上存放三份，存储策略为本地一份，同机架内其它某一节点上一份，不同机架的某一节点上一份。

4. client请求3台DataNode中的一台A上传数据（本质上是一个RPC调用，建立pipeline），A收到请求会继续调用B，然后B调用C，将整个pipeline建立完成，后逐级返回client；

5. client开始往A上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位（默认64K），A收到一个packet就会传给B，B传给C；A每传一个packet会放入一个应答队列等待应答。

6. 数据被分割成一个个packet数据包在pipeline上依次传输，在pipeline反方向上，逐个发送ack（命令正确应答），最终由pipeline中第一个DataNode节点A将pipelineack发送给client;

7. 当一个block传输完成之后，client再次请求NameNode上传第二个block到服务器。

3 HDFS文件读取过程

详细步骤解析

1. Client向NameNode发起RPC请求，来确定请求文件block所在的位置；

2. NameNode会视情况返回文件的部分或者全部block列表，对于每个block，NameNode 都会返回含有该 block 副本的 DataNode 地址； 这些返回的 DN 地址，会按照集群拓扑结构得出 DataNode 与客户端的距离，然后进行排序，排序两个规则：网络拓扑结构中距离 Client 近的排靠前；心跳机制中超时汇报的 DN 状态为 STALE，这样的排靠后；

3. Client 选取排序靠前的 DataNode 来读取 block，如果客户端本身就是DataNode,那么将从本地直接获取数据(短路读取特性)；

4. 底层上本质是建立 Socket Stream（FSDataInputStream），重复的调用父类 DataInputStream 的 read 方法，直到这个块上的数据读取完毕；

5. 当读完列表的 block 后，若文件读取还没有结束，客户端会继续向NameNode 获取下一批的 block 列表；

6. 读取完一个 block 都会进行 checksum 验证，如果读取 DataNode 时出现错误，客户端会通知 NameNode，然后再从下一个拥有该 block 副本的DataNode 继续读。

7. read 方法是并行的读取 block 信息，不是一块一块的读取；NameNode 只是返回Client请求包含块的DataNode地址，并不是返回请求块的数据；

8. 最终读取来所有的 block 会合并成一个完整的最终文件。