Flume+HBase+Kafka集成数据采集/存储/分发完整流程测试

刺骨的言语ヽ痛彻心扉 2023-02-23 08:16 6阅读 0赞

### 目录 ###

*   *  第一步：熟悉需求分析
     *  第二步：应用服务模拟器程序开发
     *   *  （1）应用服务程序开发
         *  （2）Jar包上传与分发
     *  第三步：编写运行脚本
     *   *  （1）编写运行模拟程序的Shell脚本
         *  （2）编写Flume集群服务启动脚本
         *  （3）编写Kafka Consumer执行脚本
     *  第四步：测试
     *   *  （1）启动程序
         *  （2）启动测试
         *   *  （1）通过脚本启动flume
             *  （2）测试模拟程序服务
             *  （3）测试flume与Kafka
             *  （4）测试flume与HBase

关于Flume、HBase、Kafka相关配置介绍可参考下面文章：

*  [Centos7下Kafka安装与配置][Centos7_Kafka]
 *  [Centos7下Flume安装配置与集成开发（超详细！！！）][Centos7_Flume]
 *  [Centos7下HBase安装与配置（亲测！）][Centos7_HBase]
 *  [HBase常用命令大全][HBase]

## 第一步：熟悉需求分析 ##

我们首先要了解项目需求，如图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]  
两个flume节点采集node2，node3应用服务器，汇总在node1上的flume，然后node1上flume分为两端传递数据，一端是传递给Kafka，一端是传递给HBase，之前各个组件的配置我都已经总结好，详细请看上面的几篇文章，这一篇文章主要是对整个数据采集存储、分发做一个详细的测试。

## 第二步：应用服务模拟器程序开发 ##

之前在`/opt/datas`文件夹下设置好了weblog.log日式数据文件，并且将数据的格式也已经转换好，每行数据的空格全部用逗号来隔开，但是由于原来数据的有一些中文乱码，我们就需要开发应用服务重新清洗数据，这时我们在`/opt/datas`文件夹下创建了weblog-flume.log文件，flume采集的数据就从weblog-flume.log文件中采集。这里注意的是weblog-flume.log文件均在node2和node3的节点上。

### （1）应用服务程序开发 ###

应用服务模拟器程序的代码：

import java.io.*;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/7/2 * @time : 12:40 上午 */
    public class LogReadAndWriter { 
        private static String readFileName;
        private static String writeFileName;
    
        public static void main(String[] args) { 
            readFileName = args[0];
            writeFileName = args[1];
    
            try { 
                // readInput()
                readFileByLines(readFileName);
            }catch (Exception e){ 
    
            }
        }
        public static void readFileByLines(String filename){ 
            FileInputStream fis = null;
            InputStreamReader isr = null;
            BufferedReader br = null;
            String tempString = null;
    
            try{ 
                System.out.println("以行为单位读取文件内容，一次读一整行：");
                fis = new FileInputStream(filename);
                // 从文件系统中的某个文件中获取字节
                isr = new InputStreamReader(fis,"GBK");
                br = new BufferedReader(isr);
                int count = 0;
                while ((tempString = br.readLine()) != null){ 
                    count++;
                    // 显示符号
                    Thread.sleep(100);
                    String str = new String(tempString.getBytes("UTF8"),"GBK");
                    System.out.println("row:"+count+">>>>>>>>>>"+tempString);
                    method1(writeFileName,tempString);
                }
    
            } catch (FileNotFoundException e) { 
                e.printStackTrace();
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                if (isr != null){ 
                    try{ 
                        isr.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        public static void method1(String file, String content){ 
            BufferedWriter out = null;
            try { 
                out = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file,true)));
                out.write("\n");
                out.write(content);
            } catch (FileNotFoundException e) { 
                e.printStackTrace();
            } catch (IOException e) { 
                e.printStackTrace();
            } finally { 
                try { 
                    out.close();
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

### （2）Jar包上传与分发 ###

将上面写好的代码打成jar包之后，发到node2和node3的节点上去，打成jar包方法：  
[https://blog.csdn.net/weixin\_45366499/article/details/104851893][https_blog.csdn.net_weixin_45366499_article_details_104851893]  
在opt文件夹里创建一个jars文件夹，将jar包就放入这里`/opt/jars`

## 第三步：编写运行脚本 ##

### （1）编写运行模拟程序的Shell脚本 ###

在node2和node3`/opt/datas`目录下，创建weblog-shell.sh文件，编辑内容：

# /bin/bash
    
    echo "starting log..........."
    java -jar /opt/jars/weblogs.jar /opt/datas/weblog.log /opt/datas/weblog-flume.log

注意要给shell脚本赋予权限：

chmod 777 weblog-shell.sh

### （2）编写Flume集群服务启动脚本 ###

在`/opt/Hadoop/flume`目录下，创建flume-caizhengjie-start.sh文件，在node1中编辑：

# /bin/bash
    
    echo "flume-1 start....."
    bin/flume-ng agent --conf conf -f conf/flume-conf.properties -n agent1 -Dflume.root.logger=INFO,console

然后分发到node2和node3的相同目录下  
node2:

# /bin/bash
    
    echo "flume-2 start....."
    bin/flume-ng agent --conf conf -f conf/flume-conf.properties -n agent2 -Dflume.root.logger=INFO,console

node3:

# /bin/bash
    
    echo "flume-3 start....."
    bin/flume-ng agent --conf conf -f conf/flume-conf.properties -n agent3 -Dflume.root.logger=INFO,console

注意要给shell脚本赋予权限：

chmod 777 flume-caizhengjie-start.sh

最后开启运行的命令是：

./flume-caizhengjie-start.sh

### （3）编写Kafka Consumer执行脚本 ###

在`/opt/Hadoop/kafka`目录下，创建kafka-test-consumer.sh文件，在node1中编辑：  
**注意**：这里的topic需要自己在zookeeper中创建，我这里是根据需求分析创建了weblogs

# /bin/bash
    
    echo "kafka-consumer.sh start......"
    
    bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --from-beginning --topic weblogs

然后分发到node2和node3的相同目录下  
注意要给shell脚本赋予权限：

chmod 777 kafka-test-consumer.sh

最后开启运行的命令是：

./kafka-test-consumer.sh

## 第四步：测试 ##

在进行数据的测试之前，需要启动相关程序

### （1）启动程序 ###

**启动Hadoop（node1）**

start-all.sh

**启动Zookeeper（node1，node2，node3）**

zkServer.sh start

**启动HBase（node1）**

start-hbase.sh

**启动Kafka**  
在/opt/Hadoop/kafka目录下启动  
在前台启动kafka，注意查看打印在桌面的日志，有无报错信息

bin/kafka-server-start.sh config/server.properties

如果没有报错信息，启动正常，那么就可以在后台启动了

bin/kafka-server-start.sh -daemon config/server.properties

jps查看进程  
node1:

3297 Jps
    20147 Application
    19203 ResourceManager
    20051 HRegionServer
    3268 Kafka
    18758 NameNode
    18855 DataNode
    19497 NodeManager
    19788 HMaster
    19053 SecondaryNameNode
    23343 ConsoleConsumer
    15167 QuorumPeerMain

node2:

19923 NodeManager
    19811 DataNode
    3348 Application
    20118 HRegionServer
    2699 Kafka
    10156 Jps
    16381 QuorumPeerMain
    20350 HMaster

node3:

18352 HRegionServer
    6547 Jps
    14907 QuorumPeerMain
    32205 Kafka
    18045 DataNode
    18157 NodeManager
    32319 Application

说明相关服务全部启动起来了。

### （2）启动测试 ###

#### （1）通过脚本启动flume ####

分别在每台机的在`/opt/Hadoop/flume`目录下执行：

./flume-caizhengjie-start.sh

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]  
出现这种情况，不报错，即为启动成功

#### （2）测试模拟程序服务 ####

在node2和node3`/opt/datas`目录下，执行命令：

./weblog-shell.sh

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 3]  
从结果看运行成功，数据的乱码被转换成了中文，也可以去通过查看weblog-flume.log日志文件，即数据被写入到这里面。

#### （3）测试flume与Kafka ####

当flume开启之后，需要测试kafka的消费端是否正常  
在`/opt/Hadoop/kafka`目录下，执行命令：

./kafka-test-consumer.sh

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 4]  
从运行结果来看，flume与kafka的集成也是配置成功

#### （4）测试flume与HBase ####

进入到HBase

hbase shell

查看表

list

扫描整张表

scan 'weblogs'

即可看到  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 5]  
从运行结果看，HBase也有数据，说明flume与HBase集成配置成功！

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[Centos7_Kafka]: https://blog.csdn.net/weixin_45366499/article/details/106952008
[Centos7_Flume]: https://blog.csdn.net/weixin_45366499/article/details/106988294
[Centos7_HBase]: https://blog.csdn.net/weixin_45366499/article/details/106870162
[HBase]: https://blog.csdn.net/weixin_45366499/article/details/106887526
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/202007061152504.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_weixin_45366499_article_details_104851893]: https://blog.csdn.net/weixin_45366499/article/details/104851893
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200706183404932.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2020070618344352.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200706184143319.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200706184903820.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200721143010212.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70