二、HDFS 概述与基本使用

HDFS 作为最早的大数据存储系统，存储着宝贵的数据资产，各种新的算法、框架要想得到人们的广泛使用，必须支持 HDFS 才能获取已经存储在里面的数据。所以大数据技术越发展，新技术越多，HDFS 得到的支持越多，我们越离不开 HDFS。HDFS 也许不是最好的大数据存储技术，但依然最重要的大数据存储技术。

HDFS 基本组成

1、HDFS 涉及两个重要进程：NameNode、DataNode；
2、表现形式上：主要是 目录和文件。毕竟是文件系统；
3、物理存储单元是 block。

物理存储单元为什么不是文件呢？在使用或表现形式上，HDFS的文件与Windows系统上的文件是一致的。但是考虑数据的分布式查看和计算，所以将文件内的数据分块存储是非常有必要。

HDFS 一些特点

1、一个分布式文件存储系统。可扩展性强，能存放大量数据；
2、容错率高。每一个block都有备份（包括自己，默认3个）。当主block不能出错时，可以使用备份的block；
3、适合一次写入，多次读出的场景。一个文件经过创建、写入和关闭之后就不需要改变；
4、不支持并发写入和随机写入，可以创建、删除、追加。
5、不适合大量的小文件的存储。

HDFS 的基本使用

HDFS 的基本命令与Linux操作文件或目录的命令基本是一致的：mkdir、cp、等。启动集群后，可以通过 http://flink01:9870 查看命令操作后的结果。

• 查看命令的使用帮助

[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -help rm

• 创建/删除 目录

# 创建传感器（sensor）目录
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -mkdir /sensor
# 删除目录
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -rm -r /sensor

• 上传文件：moveFromLocal、copyFromLocal、put

# moveFromLocal：从本地剪切粘贴到HDFS
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs  -moveFromLocal  ./sensor1.txt  /sensor
# copyFromLocal：从本地文件系统中拷贝文件到HDFS
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -copyFromLocal sensor2.txt /sensor
# put：等同于copyFromLocal，从本地文件系统中拷贝文件到HDFS
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -put sensor3.txt /sensor

• 从HDFS的一个路径拷贝或移动到HDFS的另一个路径：cp、mv

[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -cp /sanguo/sensor3.txt /
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -mv /sanguo/sensor3.txt /

• 下载文件：copyToLocal、get

# copyToLocal：从HDFS拷贝到本地
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -copyToLocal /sensor/sensor1.txt ./sensor
# get：等同于copyToLocal
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -get /sensor/sensor1.txt ./sensor

• 给文件追加内容：appendToFile

# appendToFile：追加一个文件到已经存在的文件末尾
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -appendToFile sensor11.txt /sensor/sensor1.txt

• 查看文件内容：cat、tail

cat：显示文件内容
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -cat /sensor/sensor1.txt

• 查看目录信息：ls

# ls: 查看目录信息
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -ls /sensor

• 设置文件的副本数量：setrep
  系统默认3个副本，设置的副本数如果大于节点数，将只有节点数的副本，后续增加节点则才能增加副本数量直至设置的副本数量。

# setrep：设置HDFS中文件的副本数量
[liuwen@flink01 hadoop-3.3.1]$ hadoop fs -setrep 10 /sensor/sensor1.txt

• 其他命令
  修改文件所属权限：-chgrp、-chmod、-chown；
  统计文件夹的大小信息：-du -s -h

HDFS Client（Java）的API

写个HDFS的Java客户端得做一些准备工作：
1、下载对应版本的依赖包。这里我没发现3.3.1的，只能使用3.1.0的。后边发现也可以；
2、配置HADOOP_HOME环境变量，对应路径就是上边依赖包的的路径；
3、环境变量path中加上%HADOOP_HOME%/bin。

• 创建Maven项目
• 添加依赖

<dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.30</version>
        </dependency>
    </dependencies>

• 在项目的resource目录下增加日志配置文件：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

• 写代码，具体就不说了
  1、连接HDFS
  2、操作
  3、关闭连接

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

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

/**
 * @author Administrator
 */
public class HDFSClient {

    FileSystem fs;

    @Before
    public void connet() throws URISyntaxException, IOException, InterruptedException {
        URI uri = new URI("hdfs://flink01:8020");
        Configuration configuration = new Configuration();
        fs = FileSystem.get(uri, configuration, "liuwen");
    }

    @After
    public void close() throws IOException {
        fs.close();
    }

    @Test
    public void testMKDir() throws URISyntaxException, IOException, InterruptedException {
        fs.mkdirs(new Path("/sensor"));
    }

    @Test
    public void testPut() throws URISyntaxException, IOException, InterruptedException {
        fs.copyFromLocalFile(new Path("D:\\hadoop\\test-data\\sensor\\input"), new Path("/sensor/input"));
    }

    @Test
    public void testGet() throws URISyntaxException, IOException, InterruptedException {
        fs.copyToLocalFile(new Path("/sensor/test.txt"), new Path("D:\\test.txt"));
    }

    @Test
    public void testListFiles() throws URISyntaxException, IOException, InterruptedException {
        RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path("/sensor"), true);

        while (iterator.hasNext()) {
            LocatedFileStatus fileStatus = iterator.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));
        }
    }
}