flume知识点

ゝ一纸荒年。 2022-02-15 17:53 190阅读 0赞

分布式日志收集系统

*  sink  
    是单线程的  
    可以使用Avro sink做汇总  
    使用sink组做单点故障和负载均衡
 *  channel：  
    可选channel:设置source属性selector.optional = channel1b  
    复用选择器：  
    [https://blog.csdn.net/weixin\_43093501/article/details/89501497][https_blog.csdn.net_weixin_43093501_article_details_89501497]
 *  agent：是一个java的进程
 *  event  
    flume事件由一个可选的header和一个二进制的body组成。  
    flume使用两个独立事务分别负责从source到channel以及channel到sink的事件组成。
 *  额外知识：

配置flume（s202）  
自定义flume的source，实现记录偏移量  
代码：  
1.查看开发者文档（不详细）  
2.查看自带的source源码编写（execsource）  
实现接口，继承父类，重写方法  
从配置文件获取属性。  
创建线程池（单个），提交任务（线程类），设置休眠。  
构造方法：  
验证文件存在？  
获取偏移量，让文件从指定偏移量读取  
run:  
读取一行，设置编码，向chancle写数据，更新偏移量并保存  
编写stop方法

打包，发到（s202）/flume/lib下  
修改集群上flume的配置文件，和自定义的属性匹配

启动flume-ng agent -f …/conf/a1.conf -n a1 -Dflume.root.logger=INFO,console

kafka channel

传统保存log的缺陷：  
内存中数据可能有堆积，溢出。  
存到文件速度慢。  
经历多个组件，效率低，出问题概率大。  
优化手段：使用kafka channel作为缓冲（效率高、数据不会丢）

补充：无source（kafka channel+sink）

修改配置文件（基于上一个配置文件修改）

注意：一点要加parseAsFlumeEvent = false  
否则，写入kafka的不是文本，而是flume的event事件对象。  
创建kafka主题  
启动kafka

自定义拦截器：  
目的：  
将接收到的数据转为json  
配置文件：

代码：  
先执行Builder内部类，然后执行自定义拦截器  
builder生命周期：构造器 -> configure -> build  
拦截器主要方法：intercept  
从事件中获取body，将数据切割，和字段（配置文件中已定义）匹配，重新填充body

[https_blog.csdn.net_weixin_43093501_article_details_89501497]: https://blog.csdn.net/weixin_43093501/article/details/89501497