Hadoop核心组件之HDFS详解【运行机制、安装、操作与实战、NameNode高可用】

╰半夏微凉° 2022-02-25 08:20 456阅读 0赞

# 一、Hadoop的3个核心组件： #

分布式文件系统：**HDFS** ——实现将文件分布式存储在很多的服务器上

分布式运算编程框架：**MapReduce** ——实现在很多机器上分布式并行运算

分布式资源调度平台：**YARN** ——帮用户调度大量的MapReduce程序，并合理分配运算资源

# 二、HDFS整体运行机制 #

HDFS：分布式文件系统

## **2.1 HDFS有着文件系统共同的特征** ##

1.  有目录结构，顶层目录是：  /
2.  系统中存放的就是文件
3.  系统可以提供对文件的：创建、删除、修改、查看、移动等功能

如：查看根目录文件

hadoop fs -ls /

## **2.2 HDFS跟普通的单机文件系统有区别** ##

1.  单机文件系统中存放的文件，是在一台机器的操作系统中
2.  hdfs的文件系统会横跨N多的机器
3.  单机文件系统中存放的文件，是在一台机器的磁盘上
4.  HDFS文件系统中存放的文件，是落在n多机器的本地单机文件系统中（HDFS是一个基于Linux本地文件系统之上的文件系统）

## **2.3 hdfs的工作机制与图解：** ##

1.  客户把一个文件存入HDFS，其实HDFS会把这个文件切块后，分散存储在N台Linux机器系统中（负责存储文件块的角色：data node）<准确来说：切块的行为是由客户端决定的>
2.  一旦文件被切块存储，那么，HDFS中就必须有一个机制，来记录用户的每一个文件的切块信息，及每一块的具体存储机器（负责记录块信息的角色是：name node）
3.  为了保证数据的安全性，hdfs可以将每一个文件块在集群中存放多个副本（到底存几个副本，是由当时存入该文件的客户端指定的）

综述：**一个hdfs系统，由一台运行了namenode的服务器，和N台运行了datanode的服务器组成！**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70][]

# 三、安装HDFS集群的具体步骤： #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 1][]

**一、首先需要准备N****台linux****服务器**

学习阶段，用虚拟机即可！

先准备4台虚拟机：1个namenode节点  + 3 个datanode 节点

**二、修改各台机器的主机名和ip****地址**

主机名：hdp-01  对应的ip地址：192.168.33.61

主机名：hdp-02  对应的ip地址：192.168.33.62

主机名：hdp-03  对应的ip地址：192.168.33.63

主机名：hdp-04  对应的ip地址：192.168.33.64

**三、从windows中用CRT软件进行远程连接**

在windows中将各台linux机器的主机名配置到的windows的本地域名映射文件中：

c:/windows/system32/drivers/etc/hosts

192.168.33.61 hdp-01
    192.168.33.62 hdp-02
    192.168.33.63 hdp-03
    192.168.33.64 hdp-04

# **四、配置Linux****服务器的基础软件环境** #

**关闭防火墙**

**CentOS6：**

service iptables stop  #关闭防火墙
    
    chkconfig iptables off #关闭防火墙自启

**CentOS7：**

firewall-cmd --state #查看默认防火墙状态（关闭后显示notrunning，开启后显示running）
    systemctl stop firewalld.service #停止firewall
    systemctl disable firewalld.service #禁止firewall开机启动

1、安装jdk：（hadoop体系中的各软件都是java开发的）

2、利用alt+p 打开sftp窗口，然后将jdk压缩包拖入sftp窗口

3、然后在linux中将jdk压缩包解压到/root/apps下

4、配置环境变量：

vi /etc/profile

在文件的最后，加入：

export JAVA_HOME=/root/apps/jdk1.8.0_60
    export PATH=$PATH:$JAVA_HOME/bin

5、修改完成后，记得执行source /etc/profile使配置生效

6、检验：在任意目录下输入命令：java -version 看是否成功执行

7、将安装好的jdk目录用scp命令拷贝到其他机器

scp -r /root/apps hdp-02:/root
    scp -r /root/apps hdp-03:/root
    scp -r /root/apps hdp-04:/root

8、将/etc/profile配置文件也用scp命令拷贝到其他机器并分别执行source命令

scp -r /etc/profile hdp-02:/etc/profile
    scp -r /etc/profile hdp-03:/etc/profile
    scp -r /etc/profile hdp-04:/etc/profile

9、集群内主机的**域名映射**配置，在hdp-01上，vi /etc/hosts

127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
    ::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
    192.168.33.61   hdp-01
    192.168.33.62   hdp-02
    192.168.33.63   hdp-03
    192.168.33.64   hdp-04

然后，将hosts文件拷贝到集群中的所有其他机器上

scp /etc/hosts hdp-02:/etc/
    
    scp /etc/hosts hdp-03:/etc/
    
    scp /etc/hosts hdp-04:/etc/

# **五、正式安装HDFS****集群** #

1、上传hadoop安装包到hdp-01

[https://hadoop.apache.org/releases.html][https_hadoop.apache.org_releases.html]

2、修改配置文件

核心配置参数：
    1) 配置JAVA环境
    2) 指定Hadoop的默认文件系统为：HDFS
    3) 指定HDFS的namenode节点为哪台机器
    4) 指定namenode软件存储元数据的本地目录
    5) 指定datanode软件存放文件块的本地目录
    6) 指定哪些机器为datanode

**3、hadoop的配置文件在：/root/apps/hadoop安装目录/etc/hadoop/**

**(1)修改hadoop-env.sh 【配置JAVA\_HOME 用于Hadoop环境 】  **

export JAVA_HOME=/root/apps/jdk1.8.0_60

**(2)修改core-site.xml 【设置默认的文件系统、设置namenode为hdp-01机器】**

<configuration>
        <property>
            <name>fs.defaultFS</name>
            <value>hdfs://hdp-01:9000</value>
        </property>
    </configuration>

**(3) 修改hdfs-site.xml 【指定namenode、datanode软件存储元数据的本地目录】**

<configuration>
        <property>
            <name>dfs.namenode.name.dir</name>
            <value>/root/hdpdata/name/</value>
        </property>
    
        <property>
            <name>dfs.datanode.data.dir</name>
            <value>/root/hdpdata/data</value>
        </property>
    
        <property>
            <name>dfs.namenode.secondary.http-address</name>
            <value>hdp-02:50090</value>
        </property>
    </configuration>

**(4) 拷贝整个hadoop安装目录到其他机器**

scp -r /root/apps/hadoop-2.8.1  hdp-02:/root/apps/
    scp -r /root/apps/hadoop-2.8.1  hdp-03:/root/apps/
    scp -r /root/apps/hadoop-2.8.1  hdp-04:/root/apps/

**(5) 启动HDFS**

所谓的启动HDFS，就是在对的机器上启动对的软件，要运行hadoop的命令，需要在linux环境中配置HADOOP\_HOME和PATH环境变量，vi /etc/profile

export JAVA_HOME=/root/apps/jdk1.8.0_60
    export HADOOP_HOME=/root/apps/hadoop-2.8.1
    export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

**首先，初始化namenode****的元数据目录**

要在hdp-01上执行hadoop的一个命令来初始化namenode的元数据存储目录

hadoop namenode -format

1.  创建一个全新的元数据存储目录
2.  生成记录元数据的文件fsimage
3.  生成集群的相关标识：如：集群id——clusterID

**然后，启动namenode****进程（在hdp-01****上）**

hadoop-daemon.sh start namenode

启动完后，首先用jps查看一下namenode的进程是否存在

**然后，在windows****中用浏览器访问namenode****提供的web****端口：50070**

http://hdp-01:50070

**然后，启动众datanode****们（在任意地方）**

hadoop-daemon.sh start datanode

**6) ****用自动批量启动脚本来启动HDFS【一个个启动datanode不累吗?】**

1.  先配置hdp-01到集群中所有机器（包含自己）的免密登陆    输入命令 ：
    
        ssh-keygen  #【一直回车】
        ssh-copy-id hdp-01   #输入密码
        ssh-copy-id hdp-02   #输入密码
        ssh-copy-id hdp-03   #输入密码
        ssh-copy-id hdp-04   #输入密码
2.  配完免密后，可以执行一次 ssh 0.0.0.0
3.  修改hadoop安装目录中/etc/hadoop/slaves（可以通过配置文件知道namenode，但无法知道datanode，所以把需要启动datanode进程的节点列入）
    
        hdp-01
        hdp-02
        hdp-03
        hdp-04
4.  在hdp-01上用脚本：**start-dfs.sh**来自动启动整个集群【这个脚本位于sbin/目录下】
5.  如果要停止，则用脚本：**stop-dfs.sh**

## 四、hdfs的客户端操作 ##

**客户端的理解：**

HDFS的客户端有多种形式：

1.  网页形式  
      
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 2][]
2.  命令行形式
3.  客户端在哪里运行，没有约束，只要运行客户端的机器能够跟HDFS集群联网

文件的**切块大小**和存储的**副本数量**，都是由客户端决定！所谓的由客户端决定，是通过配置参数来定的

hdfs的客户端会读以下两个参数，来决定切块大小、副本数量：

**切块大小的参数：dfs.blocksize**

**副本数量的参数：dfs.replication**

# 五、HDFS客户端的常用操作命令 #

查看hdfs的根目录

hadoop fs -ls /

上传文件到hdfs中【将根目录下的本地文件放入HDFS的/aaa目录下】

hadoop fs -put /本地文件 /aaa

copyFromLocal等价于put

hadoop fs -copyFromLocal /本地文件  /hdfs路径

跟copyFromLocal的区别是：从本地**移动**到hdfs中

hadoop fs -moveFromLocal /本地文件  /hdfs路径

移动HDFS中的文件（与moveFromLocal的区别是mv仅仅是HDFS文件夹的内部移动）

hadoop fs -mv /hdfs的路径1  /hdfs的另一个路径2

下载文件到客户端本地磁盘

hadoop fs -get  /hdfs中的路径   /本地磁盘目录
    hadoop fs -copyToLocal /hdfs中的路径  /本地磁盘路径   ## 跟get等价
    hadoop fs -moveToLocal  /hdfs路径  /本地路径     ## 从hdfs中移动【删除】到本地

在HDFS中创建文件夹

hadoop fs -mkdir  -p /aaa/xxx

复制HDFS中的文件到HDFS的另一个目录

hadoop fs -cp /hdfs路径_1  /hdfs路径_2

删除HDFS中的文件或文件夹

hadoop fs -rm -r /aaa

查看HDFS中的文本文件内容

hadoop fs -cat /demo.txt
    hadoop fs -tail -f /demo.txt

修改文件的权限

hadoop fs -chown user:group /aaa
    hadoop fs -chmod 700 /aaa

追加内容到已存在的文件

hadoop fs -appendToFile /本地文件   /hdfs中的文件

# 六、HDFS的JAVA客户端API使用 #

修改配置参考[http://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/hdfs-default.xml][http_hadoop.apache.org_docs_stable_hadoop-project-dist_hadoop-hdfs_hdfs-default.xml]

1、导入jar包

2、书写代码

所包含的方法有：

0、上传一个文件到HDFS中                    fs.copyFromLocalFile(...);
    1、从HDFS中下载文件到客户端本地磁盘         fs.copyToLocalFile(...);
    2、在HDFS内部移动文件\修改名称              fs.rename(...);
    3、在HDFS中创建文件夹                      fs.mkdirs(...);
    4、在HDFS中删除文件或文件夹                 fs.delete(...);
    5、查询HDFS指定目录下的文件信息             fs.listFiles(new Path("/"), true); //从返回的迭代器中取信息
    6、查询HDFS指定目录下的文件和文件夹信息      fs.listStatus(new Path("/"));
    7、读取HDFS中的文件的内容
    8、读取HDFS中文件的指定偏移量范围的内容
    9、往HDFS中的文件写内容

package cn.itcats.hdfs;
    
    public class HdfsClientDemo {
    
    	public static void main(String[] args) throws Exception {
    		/**
    		 * Configuration参数对象的机制： 构造时，会加载jar包中的默认配置 xx-default.xml 再加载
    		 * 用户配置xx-site.xml ，覆盖掉默认参数 构造完成之后，还可以conf.set("p","v")，会再次覆盖用户配置文件中的参数值
    		 */
    		// new Configuration()会从项目的classpath中加载core-default.xml hdfs-default.xml
    		// core-site.xml hdfs-site.xml等文件
    		Configuration conf = new Configuration();
    
    		// 指定本客户端上传文件到hdfs时需要保存的副本数为：2
    		conf.set("dfs.replication", "2");
    		// 指定本客户端上传文件到hdfs时切块的规格大小：64M
    		conf.set("dfs.blocksize", "64m");
    
    		// 构造一个访问指定HDFS系统的客户端对象:
    		// 参数1:——HDFS系统的URI，参数2：——客户端要特别指定的参数，参数3：客户端的身份（用户名）
    		FileSystem fs = FileSystem.get(new URI("hdfs://hdp-01:9000/"), conf, "root");
    
    		// 上传一个文件到HDFS中
    		fs.copyFromLocalFile(new Path("D:/install-pkgs/hbase-1.2.1-bin.tar.gz"), new Path("/aaa/"));
    
    		fs.close();
    	}
    
    	FileSystem fs = null;
    
    	@Before
    	public void init() throws Exception {
    		Configuration conf = new Configuration();
    		conf.set("dfs.replication", "2");
    		conf.set("dfs.blocksize", "64m");
    
    		fs = FileSystem.get(new URI("hdfs://hdp-01:9000/"), conf, "root");
    
    	}
    
    	/**
    	 * 从HDFS中下载文件到客户端本地磁盘
    	 * 
    	 * @throws IOException
    	 * @throws IllegalArgumentException
    	 */
    	@Test
    	public void testGet() throws IllegalArgumentException, IOException {
    
    		fs.copyToLocalFile(new Path("/hdp20-05.txt"), new Path("f:/"));
    		fs.close();
    
    	}
    
    	/**
    	 * 在hdfs内部移动文件\修改名称
    	 */
    	@Test
    	public void testRename() throws Exception {
    
    		fs.rename(new Path("/install.log"), new Path("/aaa/in.log"));
    
    		fs.close();
    
    	}
    
    	/**
    	 * 在hdfs中创建文件夹
    	 */
    	@Test
    	public void testMkdir() throws Exception {
    
    		fs.mkdirs(new Path("/xx/yy/zz"));
    
    		fs.close();
    	}
    
    	/**
    	 * 在hdfs中删除文件或文件夹
    	 */
    	@Test
    	public void testRm() throws Exception {
    
    		fs.delete(new Path("/aaa"), true);
    
    		fs.close();
    	}
    
    	/**
    	 * 查询hdfs指定目录下的文件信息
    	 */
    	@Test
    	public void testLs() throws Exception {
    		// 只查询文件的信息,不返回文件夹的信息
    		RemoteIterator<LocatedFileStatus> iter = fs.listFiles(new Path("/"), true);
    
    		while (iter.hasNext()) {
    			LocatedFileStatus status = iter.next();
    			System.out.println("文件全路径：" + status.getPath());
    			System.out.println("块大小：" + status.getBlockSize());
    			System.out.println("文件长度：" + status.getLen());
    			System.out.println("副本数量：" + status.getReplication());
    			System.out.println("块信息：" + Arrays.toString(status.getBlockLocations()));
    
    			System.out.println("--------------------------------");
    		}
    		fs.close();
    	}
    
    	/**
    	 * 查询hdfs指定目录下的文件和文件夹信息
    	 */
    	@Test
    	public void testLs2() throws Exception {
    		FileStatus[] listStatus = fs.listStatus(new Path("/"));
    
    		for (FileStatus status : listStatus) {
    			System.out.println("文件全路径：" + status.getPath());
    			System.out.println(status.isDirectory() ? "这是文件夹" : "这是文件");
    			System.out.println("块大小：" + status.getBlockSize());
    			System.out.println("文件长度：" + status.getLen());
    			System.out.println("副本数量：" + status.getReplication());
    
    			System.out.println("--------------------------------");
    		}
    		fs.close();
    	}
    
    	/**
    	 * 读取hdfs中的文件的内容
    	 * 
    	 * @throws IOException
    	 * @throws IllegalArgumentException
    	 */
    	@Test
    	public void testReadData() throws IllegalArgumentException, IOException {
    
    		FSDataInputStream in = fs.open(new Path("/test.txt"));
    
    		BufferedReader br = new BufferedReader(new InputStreamReader(in, "utf-8"));
    
    		String line = null;
    		while ((line = br.readLine()) != null) {
    			System.out.println(line);
    		}
    
    		br.close();
    		in.close();
    		fs.close();
    
    	}
    
    	/**
    	 * 读取hdfs中文件的指定偏移量范围的内容
    	 * 
    	 * 
    	 * 思考：用本例中的知识，实现读取一个文本文件中的指定BLOCK块中的所有数据
    	 * 
    	 * @throws IOException
    	 * @throws IllegalArgumentException
    	 */
    	@Test
    	public void testRandomReadData() throws IllegalArgumentException, IOException {
    
    		FSDataInputStream in = fs.open(new Path("/xx.dat"));
    
    		// 将读取的起始位置进行指定
    		in.seek(12);
    
    		// 读16个字节
    		byte[] buf = new byte[16];
    		in.read(buf);
    
    		System.out.println(new String(buf));
    
    		in.close();
    		fs.close();
    
    	}
    
    	/**
    	 * 往hdfs中的文件写内容
    	 * 
    	 * @throws IOException
    	 * @throws IllegalArgumentException
    	 */
    
    	@Test
    	public void testWriteData() throws IllegalArgumentException, IOException {
    
    		FSDataOutputStream out = fs.create(new Path("/zz.jpg"), false);
    
    		// D:\images\006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg
    
    		FileInputStream in = new FileInputStream("D:/images/006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg");
    
    		byte[] buf = new byte[1024];
    		int read = 0;
    		while ((read = in.read(buf)) != -1) {
    			out.write(buf,0,read);
    		}
    		
    		in.close();
    		out.close();
    		fs.close();
    
    	}
    
    }

**core-default.xml在hadoop-common包下 **

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 4][]

**hdfs-default.xml中hadoop-hdfs包下**

![20190403183650866.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 6][]

既然在**new Configuration()时**会自动从项目的classpath中加载解析core-default.xml、hdfs-default.xml、core-site.xml、hdfs-site.xml等文件，那么我们也可以直接通过修改配置文件的方式修改配置【在src下创建同名文件，如hdfs-default.xml，注意配置文件的格式】，如：

<configuration>
    	<property>
    		<name>dfs.replication</name>
    		<value>4</value>
    	</property>
    
    	<property>
    		<name>dfs.blocksize</name>
    		<value>16m</value>
    	</property>
    </configuration>

**如果即写了配置文件，又手动创建了Configuration类，并set为属性赋值，那么最终是配置文件生效还是程序手动set赋值生效呢？**

答案是：set手动赋值生效，刚才已经解释过了，先读取配置文件，后经过程序代码set赋值，set起了覆盖作用。

# 七、项目实战【日志采集系统】 #

日志采集系统架构图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 7][]

## 7.1    需求描述： ##

在业务系统的服务器上，业务程序会不断生成业务日志（比如网站的页面访问日志）

业务日志是用log4j生成的，会不断地切出日志文件

需要定期（比如每小时）从业务服务器上的日志目录中，探测需要采集的日志文件(access.log不能采)，发往HDFS

**注意点：**

（1）业务服务器可能有多台(hdfs上的文件名不能直接用日志服务器上的文件名)

（2）当天采集到的日志要放在hdfs的当天目录中

（3）采集完成的日志文件，需要移动到到日志服务器的一个备份目录中

（4）定期检查（一小时检查一次）备份目录，将备份时长超出24小时的日志文件清除

## 7.2    设计规划 ##

1、流程
    启动一个定时任务:
    	————定时探测日志源目录
    	————获取需要采集的文件
    	————移动这些文件到一个待上传目录
    	————遍历带上传目录中各文件，逐一传输到HDFS的目标路径，同时将传输完成的文件移动到备份目录
    
    启动一个定时任务：
    	————探测备份目录中的备份数据，检查是否已超出最长备份时长，如果超出，则删除
    
    2、规划各种路径
    日志源路径：d:/logs/accesslog/
    待上传临时目录: d:/logs/toupload/
    备份目录: d:/logs/backup/日期/
    
    HDFS存储路径: /logs/日期
    HDFS中的文件的前缀:access_log_
    HDFS中的文件的后缀: .log

代码如下：

public class LogCollectApp {
        public static void main(String[] args) {
            Timer timer = new Timer();
    
            //开启定时任务,延迟0ms,每隔1分钟采集一次
            timer.schedule(new CollectLogTask(), 0, 60 * 60 * 1000L);
    
            //探测备份目录中的备份数据，检查是否已超出最长备份时长，如果超出，则删除
            timer.schedule(new BackupCleanTask(), 0, 60 * 60 * 1000L);
        }
    }

public class CollectLogTask extends TimerTask {
        /**
         * ————定时探测日志源目录
         * ————获取需要采集的文件
         * ————移动这些文件到一个待上传目录
         * ————遍历带上传目录中各文件，逐一传输到HDFS的目标路径，同时将传输完成的文件移动到备份目录
         * <p>
         * 日志源路径：/Users/fatah/Desktop/logs/accesslog/
         * 待上传临时目录: /Users/fatah/Desktop/logs/toupload/
         * 备份目录: /Users/fatah/Desktop/logs/backup/日期/
         * <p>
         * HDFS存储路径: /logs/日期
         * HDFS中的文件的前缀:access_log_
         * HDFS中的文件的后缀: .log
         */
    
    
        public void run() {
            //构造一个Log4j日志对象
            Logger logger = Logger.getLogger("logRollingFile");
    
            try {
                //创建一个PropertyHolder对象(单例存在)
                Properties props = PropertyHolderLazy.getProps();
    
    
                //获取本次时间采集的日期(格式化日期)
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH");
                String day = sdf.format(new Date());
    
                File srcDir = new File(props.getProperty(Constants.LOG_SOURCE_DIR));
                //过滤名字为access.log的日志文件 因为使用滚动的日志输出 最新的日志为access.log 我们只需要access.log.x
                File[] listFiles = srcDir.listFiles(new FilenameFilter() {
                    public boolean accept(File dir, String name) {
                        if (name.startsWith(props.getProperty(Constants.LOG_LEGAL_PREFIX)))
                            return true;
                        return false;
                    }
                });
    
                //记录日志
                logger.info("探测到如下文件需要采集" + Arrays.toString(listFiles));
    
    
                //移动这些文件到一个待上传目录
                File toUploadDir = new File(props.getProperty(Constants.LOG_TOUPLOAD_DIR));
                for (File file : listFiles) {
                    FileUtils.moveFileToDirectory(file, toUploadDir, true);
                }
    
    
                //记录日志
                logger.info("如下文件移动到了待上传目录" + toUploadDir.getAbsolutePath());
    
                //创建一个HDFS客户端对象
                //遍历带上传目录中各文件，逐一传输到HDFS的目标路径(主要更名)，同时将传输完成的文件移动到备份目录
    
                FileSystem fs = FileSystem.get(new URI(props.getProperty(Constants.HDFS_URI)), new Configuration(), props.getProperty(Constants.HDFS_USERNAME));
                File[] toUploadFiles = toUploadDir.listFiles();
    
                //检查HDFS中的日期目录是否存在,如果不存在则创建
                Path hdfsDestPath = new Path(props.getProperty(Constants.HDFS_DEST_BASE_DIR) + day);
                if (!fs.exists(hdfsDestPath))
                    fs.mkdirs(hdfsDestPath);
    
    
                //检查本地的备份目录是否存在,如果不存在则创建
                File backupDir = new File(props.getProperty(Constants.LOG_BACKUP_BASE_DIR) + day + "/");
                if (!backupDir.exists())
                    backupDir.mkdirs();
    
                for (File file : toUploadFiles) {
                    //传输文件到HDFS并改名
                    Path destPath = new Path(props.getProperty(Constants.HDFS_DEST_BASE_DIR) + day + props.getProperty(Constants.HDFS_FILE_PREFIX) + UUID.randomUUID() + props.getProperty(Constants.HDFS_FILE_SUFFIX));
                    fs.copyFromLocalFile(new Path(file.getAbsolutePath()), destPath);
    
                    //记录日志
                    logger.info("文件传输到HDFS完成" + file.getAbsolutePath() + " >>> " + destPath);
    
                    //将传输完成的文件移动到备份目录
                    FileUtils.copyFileToDirectory(file, backupDir);
    
                    //记录日志
                    logger.info("文件备份完成" + file.getAbsolutePath() + " >>> " + backupDir);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
    
        }
    }

//探测备份目录中的备份数据，检查是否已超出最长备份时长，如果超出，则删除
    public class BackupCleanTask extends TimerTask {
        public void run() {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH");
            long now = new Date().getTime();
            try {
                Properties props = PropertyHolderLazy.getProps();
                //探测本地的备份目录
                File backupBaseDir = new File(props.getProperty(Constants.LOG_BACKUP_BASE_DIR));
                File[] dayBackDir = backupBaseDir.listFiles();
    
                //判断备份的子目录日期是否超过24小时
                for (File dir : dayBackDir) {
    
                    long time = sdf.parse(dir.getName()).getTime();
    
                    if (now - time > Long.parseLong(props.getProperty(Constants.LOG_BACKUP_TIMEOUT)) * 60 * 60 * 1000L) {
                        FileUtils.deleteDirectory(dir);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

public interface Constants {
    
        public static final String LOG_SOURCE_DIR = "LOG_SOURCE_DIR";
        public static final String LOG_TOUPLOAD_DIR = "LOG_TOUPLOAD_DIR";
        public static final String LOG_BACKUP_BASE_DIR = "LOG_BACKUP_BASE_DIR";
        public static final String LOG_BACKUP_TIMEOUT = "LOG_BACKUP_TIMEOUT";
        public static final String LOG_LEGAL_PREFIX = "LOG_LEGAL_PREFIX";
        public static final String HDFS_URI = "HDFS_URI";
        public static final String HDFS_DEST_BASE_DIR = "HDFS_DEST_BASE_DIR";
        public static final String HDFS_FILE_PREFIX = "HDFS_FILE_PREFIX";
        public static final String HDFS_FILE_SUFFIX = "HDFS_FILE_SUFFIX";
        public static final String HDFS_USERNAME = "HDFS_USERNAME";
    }

/**
     * 使用单例模式加载读取配置文件(饿汉式)
     */
    public class PropertyHolderEager {
        private static Properties pro = new Properties();
    
        static {
            try{
                pro.load(PropertyHolderEager.class.getResourceAsStream("collect.properties"));
            }catch (Exception e){}
        }
    
        private PropertyHolderEager() {
        }
    
        public static Properties getProps() {
            return pro;
        }
    
    }

/**
     * 使用懒汉式读取collect.properties配置文件(注意线程安全问题)
     */
    public class PropertyHolderLazy {
        private static volatile Properties pro = null;
    
        private PropertyHolderLazy() {
        }
    
        public static Properties getProps() throws IOException{
            if (pro == null) {
                synchronized (PropertyHolderLazy.class) {
                    if (pro == null) {
                        pro = new Properties();
                        pro.load(PropertyHolderLazy.class.getResourceAsStream("collect.properties"));
                    }
                }
            }
            return pro;
        }
    }

在classpath下创建：

**collect.properties**

LOG_SOURCE_DIR=/Users/fatah/Desktop/logs/accesslog/
    LOG_TOUPLOAD_DIR=/Users/fatah/Desktop/logs/toupload/
    LOG_BACKUP_BASE_DIR=/Users/fatah/Desktop/logs/backup/
    
    LOG_BACKUP_TIMEOUT=24
    
    LOG_LEGAL_PREFIX=access.log.
    
    HDFS_URI=hdfs://hdp-01:9000/
    HDFS_DEST_BASE_DIR=/logs/
    HDFS_FILE_PREFIX=access_log_
    HDFS_FILE_SUFFIX=.log
    HDFS_USERNAME=root

**log4j.properties**

### 设置###
    #log4j.rootLogger=debug,stdout,genlog
    log4j.rootLogger=INFO,logRollingFile,stdout
    
    
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.Target=System.out
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=[%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n
    
    
    ###
    log4j.logger.logRollingFile= ERROR,test1  
    log4j.appender.test1 = org.apache.log4j.RollingFileAppender 
    log4j.appender.test1.layout = org.apache.log4j.PatternLayout 
    log4j.appender.test1.layout.ConversionPattern =%d{yyyy-MMM-dd HH:mm:ss}-[TS] %p %t %c - %m%n
    log4j.appender.test1.Threshold = DEBUG 
    log4j.appender.test1.ImmediateFlush = TRUE 
    log4j.appender.test1.Append = TRUE 
    log4j.appender.test1.File = /Users/fatah/Desktop/logs/collect/collect.log 
    log4j.appender.test1.MaxFileSize = 102400KB 
    log4j.appender.test1.MaxBackupIndex = 200 
    ### log4j.appender.test1.Encoding = UTF-8

# 八、NameNode单点解决方案 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 8][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70]: /images/20220225/b556415b41784392bd71af52c321d301.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 1]: /images/20220225/065cc5b129ab447680c846829efbc7e5.png
[https_hadoop.apache.org_releases.html]: https://hadoop.apache.org/releases.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 2]: /images/20220225/e4170d9c33764a55b35eebf83e5ee425.png
[http_hadoop.apache.org_docs_stable_hadoop-project-dist_hadoop-hdfs_hdfs-default.xml]: http://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/hdfs-default.xml
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 3]: /images/20220225/c0f40220a345444b9615c5bf2c021267.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 4]: /images/20220225/94de60163fed4e6ab445e98535d1e537.png
[20190403183650866.png]: /images/20220225/5f33198a912847148c8924b6c0068fe0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 5]: /images/20220225/60653281d9e84a3dba23fb9bdf63f475.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 6]: /images/20220225/94011698c25d431d87e1b2fe9357917e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 7]: /images/20220225/98795737d3544be19f5b561ed839b894.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70 8]: /images/20220225/a52fe1ff50c247058419406bdd8b6ec3.png