hadoop(三）——hdfs（下）_current in use.lock

dfs目录：

引入：
上一篇文章中，我们看见了tmp文件下有个dfs文件夹，dfs中有存储主节点（NameNode）信息的name文件夹和存储从节点（dataNode）信息的data文件夹。接下来就详解一下这个dfs目录。

dfs的作用：
是HDFS的数据目录

dfs目录的位置：
dfs的目录是由etc/Hadoop/core-site.xml中的hadoop.tmp.dir来指定的。

dfs的目录结构：

in_use.lock：
进入tmp中的name目录或者data目录，可以看见除了current数据存储文件，还有有个in_use.lock，它的作用是，当对应进程开启的时候，HDFS对应的子目录下回出现in_use.lock。 in_use.lock标记已经开启进程，防止重复开启 。

txid：
在HDFS中，会将每一次的写操作看作是一个事务，分配一个事务id

hdfs的不可修改性：
hdfs中的文件一但上传就不可以修改，这是因为文件一但修改，另外两个备份也会要求修改，.meta数据发生改变，所以主节点的元数据也需要改变，这些工作量都太大了，所以不能修改。

edits文件和fsimage下载转化为xml的命令：

hdfs oev -i edits_xxx -o a.xml 
hdfs oiv -i fsimage_xxx -o b.xml -p XML 
1
2

从hdfs中上传/下载/删除文件的流程：

上传：

1. 客户端发起RPC请求到NameNode
2. NameNode在收到请求之后，会进行校验
3. 校验是否有写入权限
4. 校验路径下是否有同名文件
5. 如果校验失败则直接报错；如果校验成功，则NameNode会记录元数据，然后会给客户端返回一个信号表示允许上传
6. 客户端收到信号之后，会再次发送请求到NameNode，请求获取第一个Block的存储位置
7. NameNode收到请求之后，会根据副本策略等待DataNode的心跳，然后给这个Block来分配DataNode，将DataNode的地址(默认3个)放到队列中返回给客户端
8. 客户端收到队列之后，将地址从队列中取出，从这些地址中选择较近(拓扑距离)的节点将这个Block的第一个副本写入。当第一个副本写完之后，第一个副本所在的节点会通过管道(pipeline，本质上就是NIO中的Channel)将第二个副本写到对应的节点上，第二个副本写完之后，第二个副本所在的节点通过管道将第三个副本写入对应的节点上。第三个副本写完之后，第三个副本所在的节点会给第二个副本所在的节点返回一个ack信号表示写入成功；第二个副本收到ack之后会给第一个副本所在的节点返回一个ack信号；第一个副本所在的节点收到ack之后会再给客户端返回一个ack
9. 客户端在收到ack之后表示当前Block的所有副本都已经写完，那么客户端就会给NameNode发送请求来获取下一个Block的地址，重复4.5.6三步
10. 当所有的Block都写完之后，客户端就会给NameNode发送信号表示文件写完，NameNode在收到信号之后会关闭文件(实际上就是关流)
    

下载：

1. 客户端发起RPC请求到NameNode
2. NameNode在收到请求之后，会检查要读取的文件是否存在；如果不存在，则直接报错；如果存在，则会给客户端来返回一个信号表示允许读取
3. 客户端在收到信号之后，会再次给NameNode发送请求，请求获取第一个Block的存储位置
4. NameNode在收到请求之后，会查询元数据，然后将这个Block的存储位置(默认3个)放到一个队列中返回给客户端
5. 客户端在收到队列之后，将地址从队列中全部取出，从这些地址中选取一个较近的节点来读取这个Block的数据同时读取对应的meta文件，读取完成之后会对这个Block进行checksum校验(实际上就是利用meta文件校验Block)，检查校验结果，如果校验失败则客户端会通知NameNode，客户端会从剩余的地址中重选选择一个重新读取
6. 如果校验成功，则客户端会给NameNode发送请求，请求获取下一个Block的存储位置，重复3.4.5三个步骤
7. 直到读取完所有的Block，客户端会给NameNode发送结束信号。NameNode收到信号之后会关闭文件(关流)
   

删除：

1. 客户端发起RPC请求到NameNode
2. NameNode收到请求之后会将这个操作记录到edits中，记录完成之后会修改内存中的元数据，修改完成之后会给客户端返回一个ack信号表示删除成功。此时只是修改了元数据，真正的数据依然存储在DataNode上
3. NameNode会等待DataNode的心跳，收到心跳之后会校验存储信息，如果发现不一致，则会在心跳响应中要求删除对应的Block
4. DataNode在收到心跳响应之后，会删除对应的Block，此时数据才真正的从HDFS上移除

使用Java来对hdfs进行操作：

1.在pom中导入依赖：

<?xml version="1.0" encoding="UTF-8"?>
<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>cn.tedu</groupId>
    <artifactId>hdfs</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.10</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.7.1</version>
        </dependency>
    </dependencies>
</project>
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
34
35
36
37
38

2.下载文件：
注意导入的包为：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;
import java.io.*;
import java.net.URI;
1
2
3
4
5
6
7

// 下载文件
    @Test
    public void get() throws IOException {
        // 环境配置
        Configuration conf = new Configuration();
        // 连接HDFS
        FileSystem fs = FileSystem.get(
                URI.create("hdfs://10.9.162.121:9000"), // 连接地址
                conf);
        // 指定下载的文件
        InputStream in = fs.open(new Path("/a.txt"));
        // 指定写入的文件
        FileOutputStream out = new FileOutputStream("D:\\a.txt");
        // 读写文件
        IOUtils.copyBytes(in, out, conf);
        // 关流
        in.close();
        out.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

3.上传：

// 上传文件
    @Test
    public void put() throws IOException, InterruptedException {
        // 凡是可以放到XXXX-site.xml中的属性都可以在这儿配置
        Configuration conf = new Configuration();
        // conf.set("dfs.replication", "4");
        // conf.set("dfs.blocksize","56685");
        FileSystem fs = FileSystem.get(
                URI.create("hdfs://10.9.162.112:9000"),
                conf, "root");
        // 在HDFS上创建文件
        OutputStream out = fs.create(new Path("/a.log"));
        // 创建输入流
        FileInputStream in = new FileInputStream("D:\\a.txt");
        // 写出数据
        IOUtils.copyBytes(in, out, conf);
        // 关流
        in.close();
        out.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

4.删除：

 // 删除文件
    @Test
    public void delete() throws IOException, InterruptedException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(
                URI.create("hdfs://10.9.162.178:9000"),
                conf, "root");
        // 删除
        // 第二个参数表示是否允许递归删除
        fs.delete(new Path("/log"), true);
    }
1
2
3
4
5
6
7
8
9
10
11

使用Cerebrata来对hdfs文件进行操作

因为使用起来很简单，这里就直接看图说话不解释了

注意：需要在自己的host文件中添加 IP+Hadoop名称 如：192.168.0.121 Hadoop01