一家公司做两个网站吗,辅助购卡网站怎么做,wordpress文章页seo设置,ip地址信息备案管理系统文章目录 什么是FlumeFlume的特性Flume高级应用场景Flume的三大核心组件Source#xff1a;数据源channelsink Flume安装部署Flume的使用案例#xff1a;采集文件内容上传至HDFS案例#xff1a;采集网站日志上传至HDFS 各种自定义组件例如#xff1a;自定义source例如#… 文章目录 什么是FlumeFlume的特性Flume高级应用场景Flume的三大核心组件Source数据源channelsink Flume安装部署Flume的使用案例采集文件内容上传至HDFS案例采集网站日志上传至HDFS 各种自定义组件例如自定义source例如自定义sink Flume优化Flume进程监控 什么是Flume
Flume是一个高可用高可靠分布式的海量日志采集、聚合和传输的系统能够有效的收集、聚合、移动大量的日志数据。其实通俗一点来说就是Flume是一个很靠谱很方便、很强的日志采集工具。他是目前大数据领域数据采集最常用的一个框架。为什么它这么香呢 主要是因为使用Flume采集数据不需要写一行代码注意是一行代码都不需要只需要在配置文件中随便写几行配置Flume就会死心塌地的给你干活了 这个属于Flume的一个非常典型的应用场景使用Flume采集数据最终存储到HDFS上。
左边的web server表示是一个web项目web项目会产生日志数据通过中间的Agent把日志数据采集到HDFS中。其中这个Agent就是我们使用Flume启动的一个代理它是一个持续传输数据的服务数据在Agent内部的这些组件之间传输的基本单位是Event 从图中可以看到Agent是由Source、Channel、Sink这三大组件组成的这就是Flume中的三大核心组件其中source是数据源负责读取数据。channel是临时存储数据的source会把读取到的数据临时存储到channel中。sink是负责从channel中读取数据的最终将数据写出去写到指定的目的地中
Flume的特性
它有一个简单、灵活的基于流的数据流结构这个其实就是刚才说的Agent内部有三大组件数据通过这三大组件流动的具有负载均衡机制和故障转移机制一个简单可扩展的数据模型(Source、Channel、Sink)这几个组件是可灵活组合的
Flume高级应用场景
下面这个图里面主要演示了Flume的多路输出就是可以将采集到的一份数据输出到多个目的地中不同目的地的数据对应不同的业务场景 这个图里面一共有两个Agent表示我们启动了2个Flume的代理或者可以理解为了启动了2个flume的进程。首先看左边这个agent给他起个名字叫 foo这里面有一个sourcesource后面接了3个channel表示source读取到的数据会重复发送给每个channel每个channel中的数据都是一样的。针对每个channel都接了一个sink这三个sink负责读取对应channel中的数据并且把数据输出到不同的目的地
sink1负责把数据写到hdfs中 sink2负责把数据写到一个Java消息服务数据队列中 sink3负责把数据写给另一个Agent
注意了Flume中多个Agent之间是可以连通的只需要让前面Agent的sink组件把数据写到下一个Agent的source组件中即可。
所以sink3就把数据输出到了Agent bar中。在Agent bar中同样有三个组件source组件其实就获取到了sink3发送过来的数据然后把数据临时存储到自己的channel4中最终再通过sink组件把数据写到其他地方。这就是这个场景的应用把采集到的一份数据重复输出到不同的目的地中。
下面这张图这张图主要表示了flume的汇聚功能就是多个Agent采集到的数据统一汇聚到一个Agent 这个图里面一共启动了四个agent左边的三个agent都是负责采集对应web服务器中的日志数据数据采集过来之后统一发送给agent4最后agent4进行统一汇总最终写入hdfs。 这种架构的好处是后期如果要修改最终数据的输出目的地只需要修改agent4中的sink即可不需要修改agent1、2、3。但是这种架构也有弊端
如果有很多个agent同时向agent4写数据那么agent4会出现性能瓶颈导致数据处理过慢这种架构还存在单点故障问题如果agent4挂了那么所有的数据都断了。 不过这些问题可以通过flume中的负载均衡和故障转移机制解决
Flume的三大核心组件
Source数据源Channel临时存储数据的管道Sink目的地
接下来具体看一下这三大核心组件都是干什么的
Source数据源
Source数据源通过source组件可以指定让Flume读取哪里的数据然后将数据传递给后面的channel Flume内置支持读取很多种数据源基于文件、基于目录、基于TCP/UDP端口、基于HTTP、Kafka的等等、当然了如果这里面没有你喜欢的他也是支持自定义的
在这我们挑几个常用的看一下
Exec Source实现文件监控可以实时监控文件中的新增内容类似于linux中的tail -f 效果。 在这需要注意 tail -F 和 tail -f 的区别 tail -F 等同于–followname --retry根据文件名进行追踪并保持重试即该文件被删除或改名后如果再次创建相同的文件名会继续追踪 tail -f 等同于–followdescriptor根据文件描述符进行追踪当文件改名或被删除追踪停止。在实际工作中我们的日志数据一般都会通过log4j记录log4j产生的日志文件名称是固定的每天定时给文件重命名 假设默认log4j会向access.log文件中写日志每当凌晨0点的时候log4j都会对文件进行重命名在access后面添加昨天的日期然后再创建新的access.log记录当天的新增日志数据。这个时候如果想要一直监控access.log文件中的新增日志数据的话就需要使用tail -FNetCat TCP/UDP Source 采集指定端口(tcp、udp)的数据可以读取流经端口的每一行数据Spooling Directory Source采集文件夹里新增的文件Kafka Source从Kafka消息队列中采集数据
注意了前面我们分析的这几个source组件其中execsource 和 kafkasource在实际工作中是最常见的可以满足大部分的数据采集需求。
channel
Channel接受Source发出的数据可以把channel理解为一个临时存储数据的管道。 Channel的类型有很多内存、文件内存文件、JDBC等 接下来我们来分析一下
Memory Channel使用内存作为数据的存储 优点是效率高因为就不涉及磁盘IO 缺点有两个 1可能会丢数据如果Flume的agent挂了那么channel中的数据就丢失了。 2内存是有限的会存在内存不够用的情况File Channel使用文件来作为数据的存储 优点是数据不会丢失 缺点是效率相对内存来说会有点慢但是这个慢并没有我们想象中的那么慢 所以这个也是比较常用的一种channel。Spillable Memory Channel使用内存和文件作为数据存储即先把数据存到内存中如果内存中数据达到阈值再flush到文件中 优点解决了内存不够用的问题。 缺点还是存在数据丢失的风险
sink
Sink从Channel中读取数据并存储到指定目的地 Sink的表现形式有很多打印到控制台、HDFS、Kafka等
注意Channel中的数据直到进入目的地才会被删除当Sink写入目的地失败后可以自动重写 不会造成数据丢失这块是有一个事务保证的。
常用的sink组件有
Logger Sink将数据作为日志处理可以选择打印到控制台或者写到文件中这个主要在测试的时候使用HDFS Sink将数据传输到HDFS中这个是比较常见的主要针对离线计算的场景Kafka Sink将数据发送到kafka消息队列中这个也是比较常见的主要针对实时计算场景数据不落盘实时传输最后使用实时计算框架直接处理。
Flume安装部署
在这里我重新克隆了一台Linux机器主机名设置为bigdata04ip设置为192.168.182.103 关闭防火墙安装jdk并配置环境变量因为Flume是java开发所以需要依赖jdk环境。这些工作已经提前做好了继续往下面分析 想要安装Flume首先需要下载Flume进入Flume的官网找到Download链接 安装包下载好以后上传到linux机器的/data/soft目录下并且解压
[rootbigdata04 soft]# ll
total 255844
-rw-r--r--. 1 root root 67938106 May 1 23:27 apache-flume-1.9.0-bin.tar.gz
drwxr-xr-x. 7 10 143 245 Dec 16 2018 jdk1.8
-rw-r--r--. 1 root root 194042837 Apr 6 23:14 jdk-8u202-linux-x64.tar.gz
[rootbigdata04 soft]# tar -zxvf apache-flume-1.9.0-bin.tar.gz修改盘flume的env环境变量配置文件 在flume的conf目录下修改flume-env.sh.template的名字去掉后缀template
[rootbigdata04 conf]# mv flume-env.sh.template flume-env.sh 这样就好了Flume的安装是不是很简单这个时候我们不需要启动任何进程只有在配置好采集任务之后才需要启动Flume。
Flume的使用
下面我们就想上手操作Flume具体该怎么做呢 先来看一个入门级别的Hello World案例。 我们前面说了启动Flume任务其实就是启动一个AgentAgent是由source、channel、sink组成的这些组件在使用的时候只需要写几行配置就可以了 那下面我们就看一下source、channel、sink该如何配置呢接下来带着大家看一下官网找到左边的documentation查看文档信息 Flume的操作文档是非常良心的整理的非常详细
下面有一个Agent配置的例子 # example.conf: A single-node Flume configuration
# Name the components on this agent
a1.sources r1
a1.sinks k1
a1.channels c1
# Describe/configure the source
a1.sources.r1.type netcat
a1.sources.r1.bind localhost
a1.sources.r1.port 44444
# Describe the sink
a1.sinks.k1.type logger
# Use a channel which buffers events in memory
a1.channels.c1.type memory
a1.channels.c1.capacity 1000
a1.channels.c1.transactionCapacity 100
# Bind the source and sink to the channel
a1.sources.r1.channels c1
a1.sinks.k1.channel c1这个例子中首先定义了source的名字、sink的名字还有channel的名字 下面配置source的相关参数 下面配置了sink的相关参数 接着配置了channel的相关参数 最后把这三个组件连接到了一起就是告诉source需要向哪个channel写入数据告诉sink需要从哪个channel读取数据这样source、channel、sink这三个组件就联通了。总结下来配置Flume agent的主要流程是这样的
给每个组件起名字配置每个组件的相关参数把它们联通起来
注意了在Agent中配置的三大组件为什么要这样写呢如果我是第一次使用我也不会写啊。 三大组件的配置在文档中是有详细说明的来看一下在Flume Sources下面显示的都是已经内置支持的Source组件
刚才看的案例中使用的是source类型是netcat其实就是NetCat TCP Source看一下详细内容 这里面的粗体字体是必选的参数 第一个参数是为了指定source需要向哪个channel写数据这个其实是通用的参数主要看下面这三个type、bind、port
type类型需要指定为natcatbind指定当前机器的ip使用hostname也可以port指定当前机器中一个没有被使用的端口
指定bind和port表示开启监听模式监听指定ip和端口中的数据其实就是开启了一个socket的服务端等待客户端连接进来写入数据 在这里给agent起名为a1,所以netcat类型的配置如下这里面还指定了source、channel的名字并且把source和channel连接到一起了刨除这几个配置之外就剩下了三行配置就是刚才我们分析的那三个必填参数
a1.sources r1
a1.channels c1
a1.sources.r1.type netcat
a1.sources.r1.bind 0.0.0.0
a1.sources.r1.port 6666
a1.sources.r1.channels c1注意了bind参数后面指定的ip是四个0这个当前机器的通用ip因为一台机器可以有多个ip例如内网ip、外网ip如果通过bind参数指定某一个ip的话表示就只监听通过这个ip发送过来的数据了这样会有局限性所以可以指定0.0.0.0。下面几个参数都是可选配置默认可以不配置。接着是channel案例中channel使用的是memory
查看memory channel 这里面只有type是必填项其他都是可选的 最后看一下sink在案例中sink使用的是logger对应的就是Logger Sink
logger sink中默认也只需要指定type即可 后期我们如果想要使用其他的内置组件直接到官网文档这里查找即可这里面的配置有很多没有必要去记肯定记不住只要知道到哪里去找就可以。配置文件分析完了可以把这些配置放到一个配置文件中起名叫example.conf把这个配置文件放到
[rootbigdata04 ~]# cd /data/soft/apache-flume-1.9.0-bin
[rootbigdata04 apache-flume-1.9.0-bin]# cd conf/
[rootbigdata04 conf]# vi example.conf
# example.conf: A single-node Flume configuration
# Name the components on this agent
a1.sources r1
a1.sinks k1
a1.channels c1
# Describe/configure the source
a1.sources.r1.type netcat
a1.sources.r1.bind localhost
a1.sources.r1.port 44444
# Describe the sink
a1.sinks.k1.type logger
# Use a channel which buffers events in memory
a1.channels.c1.type memory
a1.channels.c1.capacity 1000
a1.channels.c1.transactionCapacity 100
# Bind the source and sink to the channel
a1.sources.r1.channels c1
a1.sinks.k1.channel c1注意了这个配置文件中的a1表示是agent的名称还有就是port指定的端口必须是未被使用的可以先查询一下当前机器使用了哪些端口端口的可用范围是1-65535如果懒得去查的话就尽量使用偏大一些的端口这样被占用的概率就非常低了。 Agent配置好了以后就可以启动了下面来看一下启动Agent的命令 可以使用命令
bin/flume-ng agent --name a1 --conf conf --conf-file example.conf -Dflume.ro 这里面使用flume-ng命令
后面指定agent表示启动一个Flume的agent代理
--name指定agent的名字
--conf指定flume配置文件的根目录
--conf-file指定Agent对应的配置文件(包含source、channel、sink配置的文件)
-D动态添加一些参数在这里是指定了flume的日志输出级别和输出位置INFO表示日志级其实agent的启动命令还可以这样写
bin/flume-ng agent -n $agent_name -c conf -f conf/flume-conf.properties.template这里面的-n属于简写完整的写法就是–name
-c完整写法的–conf
-f完整写法是–conf-file注意了由于配置文件里面指定了agent的名称为a1,所以在–name后面也需要指定a1还有就是通过–conf-file指定配置文件的时候需要指定conf目录下的example.conf配置文件
启动之后会看到如下信息表示启动成功启动成功之后这个窗口会被一直占用因为Agent服务一直在运行现在属于一个前台进程。
2020-05-02 10:14:56,464 (lifecycleSupervisor-1-4) [INFO - org.apache.flume.so 如果看到提示的有ERROR级别的日志信息就需要具体问题具体分析了一般都是配置文件配置错误了。接下来我们需要连接到source中通过netcat开启的socket服务端克隆一个bigdata04的会话因为前面启动Agent之后窗口就被占用了使用telnet命令可以连接到指定socket服务telnet后面的主机名和端口是根据example.conf配置文件中配置的
[rootbigdata04 ~]# telnet localhost 44444
-bash: telnet: command not found
[rootbigdata04 ~]# yum install -y telnet
[rootbigdata04 ~]# telnet localhost 44444
Trying ::1...
telnet: connect to address ::1: Connection refused
Trying 127.0.0.1...
Connected to localhost.
Escape character is ^].
hello world!
OK此时Flume中Agent服务是在前台运行这个服务实际工作中需要一直运行所以需要放到后台运行。 Flume自身没有提供直接把进程放到后台执行的参数所以就需要使用咱们前面学习的nohup和了。此时就不需要指定-Dflume.root.loggerINFO,console参数了默认情况下flume的日志会记录到日志文件中。停掉之前的Agent重新执行。
[rootbigdata04 apache-flume-1.9.0-bin]# nohup bin/flume-ng agent --name a1 - 启动之后通过jps命令可以查看到一个application进程这个就是启动的Agent
案例采集文件内容上传至HDFS
接下来我们来看一个工作中的典型案例 采集文件内容上传至HDFS 需求采集目录中已有的文件内容存储到HDFS 分析source是要基于目录的channel建议使用file可以保证不丢数据sink使用hdfs 下面要做的就是配置Agent了可以把example.conf拿过来修改一下新的文件名为file-to-hdfs.conf 首先是基于目录的source咱们前面说过Spooling Directory Source可以实现目录监控。来看一下这个Spooling Directory Source。 channels和type肯定是必填的还有一个是spoolDir就是指定一个监控的目录 看他下面的案例里面还多指定了一个fileHeader这个我们暂时也用不到后面等我们讲了Event之后大家就知道这个fileHeader可以干什么了先记着有这个事把。那来配置一下source
# Name the components on this agent
a1.sources r1
a1.sinks k1
a1.channels c1
# Describe/configure the source
a1.sources.r1.type spooldir
a1.sources.r1.spoolDir /data/log/studentDir
接下来是channel了 channel在这里使用基于文件的可以保证数据的安全性 如果针对采集的数据丢个一两条对整体结果影响不大只要求采集效率那么这个时候完全可以使用基于内存的channel 咱们前面的例子中使用的是基于内存的channel下面我们到文档中找一下基于文件的channel 根据这里的例子可知主要配置checkpointDir和dataDir因为这两个目录默认会在用户家目录下生成 建议修改到其他地方
checkpointDir是存放检查点目录data是存放数据的目录
a1.channels.c1.type file
a1.channels.c1.checkpointDir /data/soft/apache-flume-1.9.0-bin/data/studentDir
a1.channels.c1.dataDirs /data/soft/apache-flume-1.9.0-bin/data/studentDir/d最后是sink 因为要向hdfs中输出数据所以可以使用hdfssink hdfs.path是必填项指定hdfs上的存储目录 看这里例子中还指定了filePrefix参数这个是一个文件前缀会在hdfs上生成的文件前面加上这个前缀这个属于可选项有需求的话可以加上一般在这我们需要设置writeFormat和fileType这两个参数 默认情况下writeFormat的值是Writable建议改为Text看后面的解释如果后期想使用hive或者impala操作这份数据的话必须在生成数据之前设置为TextText表示是普通文本数据 fileType默认是SequenceFile还支持DataStream 和 CompressedStream DataStream 不会对输出数据进行压缩CompressedStream 会对输出数据进行压缩在这里我们先不使用压缩格式的所以选择DataStream 除了这些参数以外还有三个也比较重要hdfs.rollInterval、hdfs.rollSize和hdfs.rollCount
hdfs.rollInterval默认值是30单位是秒表示hdfs多长时间切分一个文件因为这个采集程序是一直运行的只要有新数据就会被采集到hdfs上面hdfs默认30秒钟切分出来一个文件如果设置为0表示不按时间切文件hdfs.rollSize默认是1024单位是字节最终hdfs上切出来的文件大小都是1024字节如果设置为0表示不按大小切文件hdfs.rollCount默认设置为10表示每隔10条数据切出来一个文件如果设置为0表示不按数据条数切文件这三个参数如果都设置的有值哪个条件先满足就按照哪个条件都会执行。在实际工作中一般会根据时间或者文件大小来切分文件我们之前在工作中是设置的时间和文件小相结合时间设置的是一小时文件大小设置的128M这两个哪个满足执行哪个所以针对hdfssink的配置最终是这样的
a1.sinks.k1.type hdfs
a1.sinks.k1.hdfs.path hdfs://192.168.182.100:9000/flume/studentDir
a1.sinks.k1.hdfs.filePrefix stu-
a1.sinks.k1.hdfs.fileType DataStream
a1.sinks.k1.hdfs.writeFormat Text
a1.sinks.k1.hdfs.rollInterval 3600
a1.sinks.k1.hdfs.rollSize 134217728
a1.sinks.k1.hdfs.rollCount 0
最后把组件连接到一起
# Bind the source and sink to the channel
a1.sources.r1.channels c1
a1.sinks.k1.channel c1把Agent的配置保存到flume的conf目录下的 file-to-hdfs.conf 文件中
[rootbigdata04 conf]# vi file-to-hdfs.conf
# Name the components on this agent
a1.sources r1
a1.sinks k1
a1.channels c1
# Describe/configure the source
a1.sources.r1.type spooldir
a1.sources.r1.spoolDir /data/log/studentDir
# Use a channel which buffers events in memory
a1.channels.c1.type file
a1.channels.c1.checkpointDir /data/soft/apache-flume-1.9.0-bin/data/student
a1.channels.c1.dataDirs /data/soft/apache-flume-1.9.0-bin/data/studentDir/d
# Describe the sink
a1.sinks.k1.type hdfs
a1.sinks.k1.hdfs.path hdfs://192.168.182.100:9000/flume/studentDir
a1.sinks.k1.hdfs.filePrefix stu-
a1.sinks.k1.hdfs.fileType DataStream
a1.sinks.k1.hdfs.writeFormat Text
a1.sinks.k1.hdfs.rollInterval 3600
a1.sinks.k1.hdfs.rollSize 134217728
a1.sinks.k1.hdfs.rollCount 0
# Bind the source and sink to the channel
a1.sources.r1.channels c1
a1.sinks.k1.channel c1下面就可以启动agent了在启动agent之前先初始化一下测试数据 创建/data/log/studentDir目录然后在里面添加一个文件class1.dat class1.dat中存储的是学生信息学生姓名、年龄、性别
[rootbigdata04 ~]# mkdir -p /data/log/studentDir
[rootbigdata04 ~]# cd /data/log/studentDir
[rootbigdata04 studentDir]# more class1.dat
jack 18 male
jessic 20 female
tom 17 male启动Hadoop集群
启动Agent使用在前台启动的方式方便观察现象
Flume怎么知道哪些文件是新文件呢它会不会重复读取同一个文件的数据呢 不会的我们到/data/log/studentDir目录看一下你就知道了
我们发现此时这个文件已经被加了一个后缀 .COMPLETED 表示这个文件已经被读取过了所以Flume在读取的时候会忽略后缀为 .COMPLETED 的文件。
案例采集网站日志上传至HDFS
需求是这样的
将A和B两台机器实时产生的日志数据汇总到机器C中通过机器C将数据统一上传至HDFS的指定目录中
注意HDFS中的目录是按天生成的每天一个目录 根据刚才的需求分析可知我们一共需要三台机器 这里使用bigdata02和bigdata03采集当前机器上产生的实时日志数据统一汇总到bigdata04机器上。其中bigdata02和bigdata03中的source使用基于file的sourceExecSource因为要实时读取文件中的新增数据 channel在这里我们使用基于内存的channel因为这里是采集网站的访问日志就算丢一两条数据对整体结果影响也不大我们只希望采集到的数据可以快读进入hdfs中所以就选择了基于内存的channel。 由于bigdata02和bigdata03的数据需要快速发送到bigdata04中为了快速发送我们可以通过网络直接传输sink建议使用avrosinkavro是一种数据序列化系统经过它序列化的数据传输起来效率更高并且它对应的还有一个avrosourceavrosink的数据可以直接发送给avrosource所以他们可以无缝衔接。 这样bigdata04的source就确定了 使用avrosource、channel还是基于内存的channelsink就使用 hdfssink因为是要向hdfs中写数据的。 这里面的组件只有execsource、avrosource、avrosink我们还没有使用过其他的组件都使用过了。最终需要在每台机器上启动一个agent启动的时候需要注意先后顺序先启动bigdata04上面的再启动bigdata02和bigdata03上面的。
具体实现这个案例
1在bigdata02上安装Flume并配置Agent
上传Flume的安装包解压
[rootbigdata02 soft]# tar -zxvf apache-flume-1.9.0-bin.tar.gz
在flume的conf目录下修改flume-env.sh.template的名字去掉后缀template
[rootbigdata02 soft]# cd apache-flume-1.9.0-bin/conf
[rootbigdata02 conf]# mv flume-env.sh.template flume-env.sh配置Agent创建文件 file-to-avro-101.conf
[rootbigdata02 conf] vi file-to-avro-101.conf
# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources r1
a1.channels c1
a1.sinks k1
# 配置source组件
a1.sources.r1.type exec
a1.sources.r1.command tail -F /data/log/access.log# 配置channel组件
a1.channels.c1.type memory
a1.channels.c1.capacity 1000
a1.channels.c1.transactionCapacity 100
# 配置sink组件
a1.sinks.k1.type avro
a1.sinks.k1.hostname 192.168.182.103
a1.sinks.k1.port 45454
# 把组件连接起来
a1.sources.r1.channels c1
a1.sinks.k1.channel c1这里面的配置没有特殊配置直接参考官网文档就可以搞定
2在bigdata03上安装Flume并配置Agent 上传Flume的安装包解压
配置Agent创建文件file-to-avro-102.conf
[rootbigdata03 conf] vi file-to-avro-102.conf
# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources r1
a1.channels c1
a1.sinks k1
# 配置source组件
a1.sources.r1.type exec
a1.sources.r1.command tail -F /data/log/access.log
# 配置channel组件
a1.channels.c1.type memory
a1.channels.c1.capacity 1000
a1.channels.c1.transactionCapacity 100
# 配置sink组件
a1.sinks.k1.type avro
a1.sinks.k1.hostname 192.168.182.103
a1.sinks.k1.port 45454
# 把组件连接起来
a1.sources.r1.channels c1
a1.sinks.k1.channel c13在bigdata04上安装Flume并配置Agent 这台机器我们已经安装过Flume了所以直接配置Agent即可 在指定Agent中sink配置的时候注意我们的需求是需要按天在hdfs中创建目录并把当天的数据上传到当天的日期目录中这也就意味着hdfssink中的path不能写死需要使用变量动态获取时间查看官方文档可知在hdfs的目录中需要使用%Y%m%d 在这还有一点需要注意的因为我们这里需要抽取时间这个时间其实是需要从数据里面抽取咱们前面说过数据的基本单位是EventEvent是一个对象后面我们会详细分析在这里大家先知道它里面包含的既有我们采集到的原始的数据还有一个header属性这个header属性是一个key-value结构的我们现在抽取时间就需要到event的header中抽取但是默认情况下event的header中是没有日期的强行抽取是会报错的会提示抽取不到返回空指针异常。 java.lang.NullPointerException: Expected timestamp in the Flume event headers, but it was null 那如何向header中添加日期呢 其实官方文档中也说了可以使用hdfs.useLocalTimeStamp或者时间拦截器时间拦截器我们后面会讲暂时最简单直接的方式就是使用hdfs.useLocalTimeStamp这个属性的值默认为false需要改为true。 配置Agent创建文件 avro-to-hdfs.conf
[rootbigdata04 conf] vi avro-to-hdfs.conf
# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources r1
a1.channels c1
a1.sinks k1
# 配置source组件
a1.sources.r1.type avro
a1.sources.r1.bind 0.0.0.0
a1.sources.r1.port 45454
# 配置channel组件
a1.channels.c1.type memory
a1.channels.c1.capacity 1000
a1.channels.c1.transactionCapacity 100
# 配置sink组件
a1.sinks.k1.type hdfs
a1.sinks.k1.hdfs.path hdfs://192.168.182.100:9000/access/%Y%m%d
a1.sinks.k1.hdfs.filePrefix access
a1.sinks.k1.hdfs.fileType DataStream
a1.sinks.k1.hdfs.writeFormat Text
a1.sinks.k1.hdfs.rollInterval 3600
a1.sinks.k1.hdfs.rollSize 134217728
a1.sinks.k1.hdfs.rollCount 0
a1.sinks.k1.hdfs.useLocalTimeStamp true
# 把组件连接起来
a1.sources.r1.channels c1
a1.sinks.k1.channel c1注意bigdata02和bigdata03中配置的a1.sinks.k1.port 的值45454需要和bigdata04中配置的三台机器中的Flume Agent都配置好了在开始启动之前需要先在bigdata02和bigdata03中生成测试数据为了模拟真实情况在这里我们就开发一个脚本定时向文件中写数据
#!/bin/bash
# 循环向文件中生成数据
while [ 1 1 ]
do
# 获取当前时间戳
curr_timedate %s
# 获取当前主机名
namehostname
echo ${name}_${curr_time} /data/log/access.log
# 暂停1秒
sleep 1
done在bigdata02和bigdata03中使用这个脚本生成数据 首先在bigdata02上创建/data/log目录然后创建 generateAccessLog.sh 脚本
[rootbigdata02 ~]# mkdir -p /data/log
[rootbigdata02 ~]# cd /data/log/
[rootbigdata02 log]# vi generateAccessLog.sh
#!/bin/bash
# 循环向文件中生成数据
while [ 1 1 ]
do
# 获取当前时间戳
curr_timedate %s
# 获取当前主机名
namehostname
echo ${name}_${curr_time} /data/log/access.log
# 暂停1秒
sleep 1
done接着在bigdata03上创建/data/log目录然后创建 generateAccessLog.sh 脚本
[rootbigdata03 ~]# mkdir /data/log
[rootbigdata03 ~]# cd /data/log/
[rootbigdata03 log]# vi generateAccessLog.sh
#!/bin/bash
# 循环向文件中生成数据
while [ 1 1 ]
do
# 获取当前时间戳
curr_timedate %s
# 获取当前主机名
namehostname
echo ${name}_${curr_time} /data/log/access.log
# 暂停1秒
sleep 1
done接下来开始启动相关的服务进程 首先启动bigdata04上的agent服务 接下来启动bigdata-02上的agent服务和shell脚本
[rootbigdata02 apache-flume-1.9.0-bin]# bin/flume-ng agent --name a1 --conf
[rootbigdata02 log]# sh -x generateAccessLog.sh最后启动bigdata-03上的agent服务和shell脚本
[rootbigdata03 apache-flume-1.9.0-bin]# bin/flume-ng agent --name a1 --conf
[rootbigdata03 log]# sh -x generateAccessLog.sh验证结果查看hdfs上的结果数据在bigdata01上查看
[rootbigdata01 soft]# hdfs dfs -cat /access/20200502/access.1588426157482.tmp
bigdata02_1588426253
bigdata02_1588426254
bigdata02_1588426255
bigdata02_1588426256
bigdata02_1588426257
bigdata02_1588426258注意启动之后稍等一会就可以看到数据了我们观察数据的变化会发现hdfs中数据增长的不是很快它会每隔一段时间添加一批数据实时性好像没那么高
这是因为avrosink中有一个配置batch-size它的默认值是100也就是每次发送100条数据如果数据不够100条则不发送。 具体这个值设置多少合适要看你source数据源大致每秒产生多少数据以及你希望的延迟要达到什么程度如果这个值设置太小的话会造成sink频繁向外面写数据这样也会影响性能。最终依次停止bigdata02、bigdata03中的服务最后停止bigdata04中的服务
各种自定义组件
咱们前面讲了很多组件有核心组件和高级组件 source、channel、sink以及Source InterceptorsChannel Selectors、Sink Processors 针对这些组件Flume都内置提供了组件的很多具体实现在实际工作中95%以上的数据采集需求都是可以满足的但是谁也不敢保证100%都能满足因为什么奇葩的需求都会有那针对系统内没有提供的一些组件怎么办呢 假设我们想把flume采集到的数据输出到mysql中那这个时候就需要有针对mysql的sink组件了但是Flume中并没有因为这种需求不常见往mysql中写的都是结构化数据数据的格式是固定的但是flume采集的一般都是日志数据这种属于非结构化数据不支持也是正常的但是我们在这里就是需要使用Flume往mysql中写数据那怎么办 要不我们考虑换一个采集工具把当然这也是一种解决方案如果有其他采集工具支持向mysql中写数据的话那可以考虑换一个采集工具如果所有的采集工具都不支持向mysql中写数据呢也就是说你这个需求就是前无古人后无来者的怎么破 不用担心天无绝人之路其实咱们使用的Flume提供的那些内置组件也都是作者一行代码一行代码写出来的那我们是不是也可以自己写一个自定义的组件呢可以的并且flume也很欢迎你这样去做它把开发文档什么的东西都给你准备好了。 注意了就算没有文档我们也要想办法去自定义没有文档的话就需要去抠Flume的源码了。 在这里Flume针对自定义组件提供了详细的文档说明我们来看一下通过Flume User Guide可以看到针对source、channle、sink、Source InterceptorsChannel Selectors、都是可以的这里面都显示了针对自定义的组件如何配置使用Sink Processors目前暂时不支持自定义。 那这些支持自定义的组件具体开发步骤是什么样的呢代码该写成什么样的呢大家还记得Flume有两个文档链接吗Flume Developer Guide 例如自定义source 例如自定义sink 自定义channel的内容目前还没完善如果你确实想自定义这个组件就需要到Flume源码中找到目前支持的那些channel的代码参考着实现我们自定义的channel组件。 大家在这里知道可以自定义并且知道自定义组件的文档在哪里就可以了目前来说需要我们自定义组件的场景实在是太少了几乎和买彩票中奖的概率差不多。 前面我们掌握了Flume的基本使用和高级使用场景下面我们来看一下针对Flume的一些企业级优化和监控手段
Flume优化
调整Flume进程的内存大小建议设置1G~2G太小的话会导致频繁GC 因为Flume进程也是基于Java的所以就涉及到进程的内存设置一般建议启动的单个Flume进程(或者说单个Agent)内存设置为1G~2G内存太小的话会频繁GC影响Agent的执行效率。
那具体设置多少合适呢 这个需求需要根据Agent读取的数据量的大小和速度有关系所以需要具体情况具体分析当Flume的Agent启动之后对应就会启动一个进程我们可以通过jstat -gcutil PID 1000来看看这个进程GC的信息每一秒钟刷新一次如果GC次数增长过快说明内存不够用。使用jps查看目前启动flume进程
[rootbigdata04 ~]# jps
2957 Jps
2799 Application执行 jstat -gcutil PID 1000
[rootbigdata04 ~]# jstat -gcutil 2799 1000
S0 S1 E O M CCS YGC YGCT FGC FGCT GCT
100.00 0.00 17.54 42.80 96.46 92.38 8 0.029 0 0.000 0
100.00 0.00 17.54 42.80 96.46 92.38 8 0.029 0 0.000 0
100.00 0.00 17.54 42.80 96.46 92.38 8 0.029 0 0.000 0
100.00 0.00 17.54 42.80 96.46 92.38 8 0.029 0 0.000 0
100.00 0.00 17.54 42.80 96.46 92.38 8 0.029 0 0.000 0
100.00 0.00 17.54 42.80 96.46 92.38 8 0.029 0 0.000 0在这里主要看YGC YGCT FGC FGCT GCT
YGC表示新生代堆内存GC的次数如果每隔几十秒产生一次也还可以接受如果每秒都会发生一次YGC那说明需要增加内存了YGCT表示新生代堆内存GC消耗的总时间FGCFULL GC发生的次数注意如果发生FUCC GC则Flume进程会进入暂停状态FUCC GC执行完以后Flume才会继续工作所以FUCC GC是非常影响效率的这个指标的值越低越好没有更好。GCT所有类型的GC消耗的总时间
如果需要调整Flume进程内存的话需要调整 flume-env.s h脚本中的 JAVA_OPTS 参数把 export JAVA_OPTS 参数前面的#号去掉才会生效。
export JAVA_OPTS-Xms1024m -Xmx1024m -Dcom.sun.management.jmxremote 建议这里的 Xms 和 Xmx 设置为一样大避免进行内存交换内存交换也比较消耗性能。
在一台服务器启动多个agent的时候建议修改配置区分日志文件
因为在conf目录下有log4j.properties在这里面指定了日志文件的名称和位置所有使用conf目录下面配置启动的Agent产生的日志都会记录到同一个日志文件中如果我们在一台机器上启动了10几个Agent后期发现某一个Agent挂了想要查看日志分析问题这个时候就疯了因为所有Agent产生的日志都混到一块了压根都没法分析日志了。 所以建议拷贝多个conf目录然后修改对应conf目录中log4j.properties日志的文件名称(可以保证多个agent的日志分别存储)并且把日志级别调整为warn(减少垃圾日志的产生)默认info级别会记录很多日志信息。这样在启动Agent的时候分别通过–conf参数指定不同的conf目录后期分析日志就方便了每一个Agent都有一个单独的日志文件。
以bigdata04机器为例 复制conf-failover目录以后启动sink的failover任务的时候使用这个目录 修改 log4j.properties中的日志记录级别和日志文件名称日志文件目录可以不用修改统一使用logs目录即可。
[rootbigdata04 apache-flume-1.9.0-bin]# cp -r conf/ conf-failover
[rootbigdata04 apache-flume-1.9.0-bin]# cd conf-failover/
[rootbigdata04 conf-failover]# vi log4j.properties
.....
flume.root.loggerWARN,LOGFILE
flume.log.dir./logs
flume.log.fileflume-failover.log再启动的时候就是这样的了
[rootbigdata04 apache-flume-1.9.0-bin]# nohup bin/flume-ng agent --name a1 - 这样就会在flume的logs目录中产生 flume-failover.log 文件并且文件中只记录WARN和ERROR级别 的日志这样后期排查日志就很清晰了。
[rootbigdata04 apache-flume-1.9.0-bin]# cd logs/
[rootbigdata04 logs]# ll
total 4
-rw-r--r--. 1 root root 478 May 3 16:25 flume-failover.log
[rootbigdata04 logs]# more flume-failover.log
03 May 2020 16:25:38,992 ERROR [SinkRunner-PollingRunner-FailoverSinkP
rocessor] (org.apache.flume.SinkRunner$PollingRunner.run:158) - Unabl
e to deliver event. Exception follows.
org.apache.flume.EventDeliveryException: All sinks failed to process,
nothing left to failover to
at org.apache.flume.sink.FailoverSinkProcessor.process(Failove
rSinkProcessor.java:194)
at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.ja
va:145)
at java.lang.Thread.run(Thread.java:748)Flume进程监控
Flume的Agent服务是一个独立的进程假设我们使用source-channel-sink实现了一个数据采集落盘的功能如果这个采集进程被误操作干掉了这个时候我们是发现不了的什么时候会发现呢 可能第二天产品经理找到你了说昨天的这个指标值有点偏低啊你来看下怎么回事然后你就一顿操作猛如虎结果发现原始数据少了一半多那是因为Flume的采集程序在昨天下午的时候被误操作干掉了。 找到问题之后你就苦巴巴的手工去补数据重跑计算程序最后再找产品经理确认数据的准确性。类似的问题会有很多这说明你现在是无法掌控你手下的这些程序他们都是不受控的状态说不定哪天哪个程序不高兴他就自杀了不干活了过了好几天需要用到这个数据的时候你才发现发现的早的话还能补数据发现晚的话数据可能都补不回来了这样对公司来说就是属于比较严重的数据故障问题这样你年终奖想拿18薪就不太现实了。 所以针对这些存在单点故障的进程我们都需要添加监控告警机制最起码出问题能及时知道再好一点的呢可以尝试自动修复重启。 那针对Flume中的Agent我们就来实现一个监控功能并且尝试自动重启 大致思路是这样的
首先需要有一个配置文件配置文件中指定你现在需要监控哪些Agent有一个脚本负责读取配置文件中的内容定时挨个检查Agent对应的进程还在不在如果发现对应的进程不在则记录错误信息然后告警(发短信或者发邮件) 并尝试重启 创建一个文件 monlist.conf文件中的第一列指定一个Agent的唯一标识后期需要根据这个标识过滤对应的Flume进程所以一定要保证至少在一台机器上是唯一的 等号后面是一个启动Flume进程的脚本这个脚本和Agent的唯一标识是一一对应的后期如果根据 Agent标识没有找到对应的进程那么就需要根据这个脚本启动进程
examplestartExample.sh 这个脚本的内容如下 startExample.sh
#!/bin/bash
flume_path/data/soft/apache-flume-1.9.0-bin
nohup ${flume_path}/bin/flume-ng agent --name a1 --conf ${flume_path}/conf/ -接着就是要写一个脚本来检查进程在不在不在的话尝试重启 创建脚本 monlist.sh
#!/bin/bash
monlistcat monlist.conf
echo start check
for item in ${monlist}
do
# 设置字段分隔符
OLD_IFS$IFS
IFS
# 把一行内容转成多列[数组]
arr($item)
# 获取等号左边的内容
name${arr[0]}
# 获取等号右边的内容
script${arr[1]}
echo time is:date %Y-%m-%d %H:%M:%S check $name
if [ jps -m|grep $name | wc -l -eq 0 ]
then
# 发短信或者邮件告警
echo date %Y-%m-%d %H:%M:%S$name is none
sh -x ./${script}
fi
done注意这个需要定时执行所以可以使用crontab定时调度
* * * * * root /bin/bash /data/soft/monlist.sh
