盛日辉的博客
0%

Hadoop-HDFS

发表于 更新于 分类于 ，

HDFS 的 Shell 操作

基本格式

hadoop fs ...，或 hdfs dfs ...。
后面的命令用法和Linux命令操作用法基本差不多。

常用命令

命令 作用 示例 说明
hadoop fs -mkdir 文件名 创建文件夹 hadoop fs -mkdir /xiyou/bajie 创建/xiyou/bajie的文件夹
hadoop fs -moveFromLocal 剪切本地文件上传倒hdfs流经 hadoop fs -moveFromLocal a.txt /xiyou 将本地的a.txt剪切掉，上传到/xiyou目录下
hadoop fs -copyFromLocal 拷贝本地文件上传倒hdfs流经 hadoop fs -copyFromLocal b.txt /xiyou 将本地的b.txt上传到/xiyou目录下，本地文件还在
hadoop fs -put 同-copyFromLocal hadoop fs -put wukong.txt /xiyou 将本地的wukong.txt上传到/xiyou目录下，本地文件还在
hadoop fs -appendToFile 追加一个文件到已经存在的文件末尾 hadoop fs -appendToFile jingubnag.txt /xiyou/wukong.txt 将jingubang.txt里的内容追加到wukong.txt末尾
hadoop fs -copyToLocal 从HDFS拷贝到本地 hadoop fs -copyToLocal /xiyou/a.txt ./ 把a.txt下载到本地
hadoop fs -get 同-CopyToLocal 略 略

其他命令

  1. -ls: 显示目录信息
  2. -cat：显示文件内容
  3. -chgrp、-chmod、-chown：Linux文件系统中的用法一样，修改文件所属权限
  4. -cp：从HDFS的一个路径拷贝到HDFS的另一个路径
  5. -mv：在HDFS目录中移动文件
  6. -tail：显示一个文件的末尾1kb的数据、
  7. -rm：删除文件或文件夹
  8. -rm -r：递归删除目录及目录里面内容
  9. -du统计文件夹的大小信息
  10. -setrep：设置HDFS中文件的副本数量

HDFS 的 API 操作

Maven 和 idea 的安装与配置

  1. 下载maven，防灾一个没有中文没有空格的路径下；
  2. 配置maven环境变量；
  3. 修改本地仓库路径和镜像（要进入setting.xml去修改！！）；
  4. 安装idea，并修改maven配置，在File-New Projects Setup-Settings for New Projects...里进行全局修改。

客户端代码操作常用套路

  1. 获取客户端对象
  2. 执行操作命令
  3. 关闭资源

以创建文件夹为例

  1. 先创建一个maven项目；
  2. 添加相关依赖坐标： org.apache.hadoop hadoop-client 3.1.3 junit junit 4.12 org.slf4j slf4j-log4j12 1.7.30
  3. 日志添加。在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入
    log4j.rootLogger=INFO, stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
    log4j.appender.logfile=org.apache.log4j.FileAppender
    log4j.appender.logfile.File=target/spring.log
    log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
    log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
  4. 在 /src/man/java下创建一个 com.shengrihui.hdfs 包；
  5. 创建 HDFSC列宁他 文件
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
39
40
41
42
43
44
45
46
47
48
49
50
package com.shengrihui.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * 客户端代码操作常用套路
 * 1.获取客户端对象
 * 2.执行操作命令
 * 3.关闭资源
 */

public class HdfsClient {

    private FileSystem fs;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        // 链接集群的nn地址
        URI uri = new URI("hdfs://hadoop102:8020");
        // 创建一个配置文件
        Configuration configuration = new Configuration();
        // 用户
        String user = "shengrihui";
        // 1.获取到了客户端对象
        fs = FileSystem.get(uri, configuration, user);
    }

    @After
    public void close() throws IOException {
        // 3.关闭资源
        fs.close();
    }

    @Test
    public void testMkdirs() throws URISyntaxException, IOException, InterruptedException {

        // 2.创建一个文件夹
        fs.mkdirs(new Path("/xiyou/wukong"));

    }
}

close(),init()是为了封装代码。
运行后结果：
创建文件夹

上传

1
2
3
4
5
6
7
8
9
10
11
@Test
public void testPut() throws IOException {
    /**
     * 参数：
     * 参数1：是否删除源文件
     * 参数2：是否覆盖目标文件（如果已经有了）
     * 参数3：源数据路径
     * 参数4：目的地路径
     */
    fs.copyFromLocalFile(false, true, new Path("E:\\CS\\BigData\\hadoop\\笔记（word版本）\\笔记（word版本）"), new Path("/"));
}

运行结果：
上传代码运行结果
上传结果

参数优先级

参数优先级排序：
（1）客户端代码中设置的值 >
在init里configuration.set("dfs.replication","3");
（2）ClassPath下的用户自定义配置文件 >
在resource目录下新建文件hdfs-site.xml，写入内容；

1
2
3
4
5
6
7
8
9
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<configuration>
<property>
	<name>dfs.replication</name>
     <value>1</value>
</property>
</configuration>

（3）然后是服务器的自义配置（xxx-site.xml） >
（4）服务器的默认配置（xxx-default.xml）

文件下载

1
2
3
4
5
6
7
8
9
10
11
//文件下载
@Test
public void testGet() throws IOException {
    /**
     * 参数1：源文件是否删除；参数2：源文件的路径（hdfs路径）；参数3：目标地址路径（Windows）
     * 参数4：是否文件校验，从hdfs下载来的数据可能会丢一些数据或错误，会带一个crc的校验码，
     * 再用crc算法算一个值，如果算出来的crc和传过来crc一样，就说明传来的一样
     * false是进行校验，一般不用这个参数
     */
    fs.copyToLocalFile(false, new Path("hdfs://hadoop102/xiyouji"), new Path("E:\\"), false);
}

运行结果：略

删除

1
2
3
4
5
6
7
8
9
10
11
12
//删除
@Test
public void testRm() throws IOException {
    //参数1：要删除的路径；参数2：是否递归删除
    //删除文件

    //删除空目录

    //删除非空目录（必须递归）
    //fs.delete(new Path("hdfs://hadoop102/jinguo"), true);
    fs.delete(new Path("hdfs://hadoop102/xiyou"),true);
}

运行结果：略

文件的更名和移动

1
2
3
4
5
6
7
8
9
10
11
12
13
//文件的更名和移动
    @Test
    public void testmv() throws IOException {
        //参数解读：参数1，原文件路径；参数2，目标文件路径
        //文件名称修改
        //fs.rename(new Path("hdfs://hadoop102/笔记（word版本）"),new Path("hdfs://hadoop102/note"));

        //文件更名移动
        //fs.rename(new Path("/wcinput/word.txt"),new Path("/sanguo/a.txt"));

        //目录更名
        fs.rename(new Path("/sanguo"),new Path("/shuihu"));
    }

获取文件详细信息

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
// 获取文件详细信息
@Test
public void fileDetail() throws IOException {
    // 获取所有文件信息
    RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

    // 遍历文件
    while (listFiles.hasNext()) {
        LocatedFileStatus fileStatus = listFiles.next();

        System.out.println("=======" + fileStatus.getPath() + "=======");
        System.out.println(fileStatus.getPermission());// 权限
        System.out.println(fileStatus.getOwner());
        System.out.println(fileStatus.getGroup());
        System.out.println(fileStatus.getLen());
        System.out.println(fileStatus.getModificationTime());// 上次修改的时间
        System.out.println(fileStatus.getReplication());
        System.out.println(fileStatus.getBlockSize());
        System.out.println(fileStatus.getPath().getName());

        // 获取块信息
        BlockLocation[] blockLocations = fileStatus.getBlockLocations();
        System.out.println(Arrays.toString(blockLocations));
    }
}

运行结果：
hdfs获取文件信息
其中，大文件会按快大小128M分成好几个块；
hdfs大文件块信息

判断是文件夹还是文件

1
2
3
4
5
6
7
8
9
10
11
12
13
// 判断是文件夹还是文件
@Test
public void testFile() throws IOException {
    FileStatus[] listStatus = fs.listStatus(new Path("/"));

    for (FileStatus status : listStatus) {
        if (status.isFile()) {
            System.out.println("文件 " + status.getPath().getName());
        } else {
            System.out.println("目录 " + status.getPath().getName());
        }
    }
}

HDFS读写流程

HDFS 写流程

HDFS读写流程-总
（1）客户端通过Distributed FileSystem模块向NameNode请求上传文件，NameNode检查目标文件是否已存在，父目录是否存在。
还有是否有权限上传
（2）NameNode返回是否可以上传。
（3）客户端请求第一个 Block上传到哪几个DataNode服务器上。
（4）NameNode返回3个DataNode节点，分别为dn1、dn2、dn3。
节点的选择涉及到节点距离最近、负载均衡（如果多个客户端都在写，那就pass这个节点）
（5）客户端通过FSDataOutputStream模块请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成。
（6）dn1、dn2、dn3逐级应答客户端。
（7）客户端开始往dn1上传第一个Block（先从磁盘读取数据放到一个本地内存缓存），以Packet为单位，dn1收到一个Packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个应答队列等待应答。
（8）当一个Block传输完成之后，客户端再次请求NameNode上传第二个Block的服务器。（重复执行3-7步）。
为什么不是多个节点同时连接客户端？防止因为有任何一个节点没有完成而一直在等；
到dn1的时候同时就往后倒dn2；
最小单位是516字节（4个校验位），攒倒64kB的packet发送；
ACK队列，接受下一端是否接受成功，用于备份。

网络拓扑-节点距离计算

节点距离：两个节点到达最近的共同祖先的距离总和。
HDFS读写数据-节点距离

机架感知（副本存储节点选择）

HDFS读写流程-机架感知
第一个副本：本地（如果客户端不在集群，随机算一个）
本地是为了上传的速度
第二个副本：另一个机架里随机
另一个机架是为了数据的可靠性
第三个副本：和第二个同一个机架随机
兼顾可靠性和速度

HDFS读数据流程

HDFS读写流程-读数据
（1）客户端通过DistributedFileSystem向NameNode请求下载文件，NameNode通过查询元数据，找到文件块所在的DataNode地址。
（2）挑选一台DataNode（就近原则，然后随机）服务器，请求读取数据。
节点距离最近，也考虑负载
（3）DataNode开始传输数据给客户端（从磁盘里面读取数据输入流，以Packet为单位来做校验）。
（4）客户端以Packet为单位接收，先在本地缓存，然后写入目标文件。
串行读，先读第一块，再读第二块

其他（了解和理解）

  • NameNode和SecondaryNameNode
    • NN和2NN工作机制
    • Fsimage和Edits解析
    • CheckPoint时间设置
  • DataNode
    • DataNode工作机制
    • 数据完整性
    • 掉线时限参数设置