Flume
概述
Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统。Flume基于流式架构，灵活简单。php
主要做用：实时读取服务器本地磁盘数据，将数据写入HDFS；html
优势：前端
- 能够和任意存储进程集成。
- 输入的的数据速率大于写入目的存储的速率（读写速率不一样步），flume会进行缓冲，减少hdfs的压力。
- flume中的事务基于channel，使用了两个事务模型（sender + receiver），确保消息被可靠发送。
Flume使用两个独立的事务分别负责从soucrce到channel，以及从channel到sink的事件传递。一旦事务中全部的数据所有成功提交到channel，那么source才认为该数据读取完成。同理，只有成功被sink写出去的数据，才会从channel中移除；失败后就从新提交；java
组成：Agent 由 source+channel+sink构成；linux
source是数据来源的抽象，sink是数据去向的抽象；web
Source
Source是负责接收数据到Flume Agent的组件。Source组件能够处理各类类型、各类格式的日志数据
数据输入端输入类型：spooling directory（spooldir）文件夹里边的数据不停的滚动、exec 命令的执行结果被采集
syslog系统日志、avro上一层的flume、netcat网络端传输的数据shell
Channel
Channel是位于Source和Sink之间的缓冲区。所以，Channel容许Source和Sink运做在不一样的速率上。Channel是线程安全的，能够同时处理几个Source的写入操做和几个Sink的读取操做。
Flume自带两种Channel：Memory Channel和File Channel。
Memory Channel是内存中的队列。Memory Channel在不须要关心数据丢失的情景下适用。若是须要关心数据丢失，那么Memory Channel就不该该使用，由于程序死亡、机器宕机或者重启都会致使数据丢失。
File Channel将全部事件写到磁盘。所以在程序关闭或机器宕机的状况下不会丢失数据。apache
Channel选择器是决定Source接收的一个特定事件写入哪些Channel的组件，它们告知Channel处理器，而后由其将事件写入到每一个Channel。vim
Channel Selector有两种类型：Replicating Channel Selector（default，会把全部的数据发给全部的Channel）和Multiplexing Chanell Selector（选择把哪一个数据发到哪一个channel）和自定义选择器；浏览器
Source 发送的 Event 经过 Channel 选择器来选择以哪一种方式写入到 Channel 中，Flume 提供三种类型 Channel 选择器，分别是复制、复用和自定义选择器。
- 复制选择器: 一个 Source 以复制的方式将一个 Event 同时写入到多个 Channel 中，不一样的 Sink 能够从不一样的 Channel 中获取相同的 Event，好比一份日志数据同时写 Kafka 和 HDFS，一个 Event 同时写入两个 Channel，而后不一样类型的 Sink 发送到不一样的外部存储。
-
该选择器复制每一个事件到经过Source的channels参数所指定的全部的Channels中。复制Channel选择器还有一个可选参数optional，该参数是空格分隔的channel名字列表。此参数指定的全部channel都认为是可选的，因此若是事件写入这些channel时，如有失败发生，会忽略。而写入其余channel失败时会抛出异常。
2. （多路）复用选择器: 须要和拦截器配合使用，根据 Event 的头信息中不一样键值数据来判断 Event 应该写入哪一个 Channel 中。
还有一种是kafka channel，它是没有sink；
3. 自定义选择器
Sink
数据去向常见的目的地有：HDFS、Kafka、logger（记录INFO级别的日志）、avro（下一层的Flume）、File、Hbase、solr、ipc、thrift自定义等
Sink不断地轮询Channel中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另外一个Flume Agent。
Sink是彻底事务性的。在从Channel批量删除数据以前，每一个Sink用Channel启动一个事务。批量事件一旦成功写出到存储系统或下一个Flume Agent，Sink就利用Channel提交事务。事务一旦被提交，该Channel从本身的内部缓冲区删除事件。
Sink groups容许组织多个sink到一个实体上。 Sink processors（处理器）可以提供在组内全部Sink之间实现负载均衡的能力，并且在失败的状况下可以进行故障转移从一个Sink到另外一个Sink。简单的说就是一个source 对应一个Sinkgroups，即多个sink,这里实际上复用/复制状况差很少，只是这里考虑的是可靠性与性能，即故障转移与负载均衡的设置。
DefaultSink Processor 接收单一的Sink，不强制用户为Sink建立Processor
FailoverSink Processor故障转移处理器会经过配置维护了一个优先级列表。保证每个有效的事件都会被处理。
工做原理是将连续失败sink分配到一个池中，在那里被分配一个冷冻期，在这个冷冻期里，这个sink不会作任何事。一旦sink成功发送一个event，sink将被还原到live 池中。
Load balancing Processor负载均衡处理器提供在多个Sink之间负载平衡的能力。实现支持经过① round_robin（轮询）或者② random（随机）参数来实现负载分发，默认状况下使用round_robin
事务
Put事务流程：
doPut将批数据先写入临时缓冲区putList； doCommit：检查channel内存队列是否足够合并； doRollback：channel内存队列空间不足，回滚数据；
尝试put先把数据put到putList里边，而后commit提交，查看channel中事务是否提交成功，若是都提交成功了就把这个事件从putList中拿出来；若是失败就重写提交，rollTback到putList；
Take事务：
doTake先将数据取到临时缓冲区takeList； doCommit若是数据所有发送成功，则清除临时缓冲区takeList； doRollback数据发送过程当中若是出现异常，rollback将临时缓存takeList中的数据归还给channel内存队列；
拉取事件到takeList中，尝试提交，若是提交成功就把takeList中数据清除掉；若是提交失败就重写提交，返回到channel后重写提交；
这种事务：flume有可能有重复的数据；
Event
传输单元，Flume数据传输的基本单元，以事件的形式将数据从源头送至目的地。 Event由可选的header和载有数据的一个byte array 构成。Header是容纳了key-value字符串对的HashMap。
拦截器（interceptor）
拦截器是简单插件式组件，设置在Source和Source写入数据的Channel之间。每一个拦截器实例只处理同一个Source接收到的事件。
由于拦截器必须在事件写入channel以前完成转换操做，只有当拦截器已成功转换事件后，channel（和任何其余可能产生超时的source）才会响应发送事件的客户端或sink。
Flume官方提供了一些经常使用的拦截器，也能够自定义拦截器对日志进行处理。自定义拦截器只需如下几步：
- 使用的Flume版本为：apache-flume-1.6.0
实现org.apache.flume.interceptor.Interceptor接口
Flume拓扑结构
① 串联：channel多，但flume层数不宜过多；这种模式是将多个flume给顺序链接起来了，从最初的source开始到最终sink传送的目的存储系统。此模式不建议桥接过多的flume数量， flume数量过多不只会影响传输速率，并且一旦传输过程当中某个节点flume宕机，会影响整个传输系统。
② 单source，多channel、sink； 一个channel对应多个sink； 多个channel对应多个sink；
---->sink1 ---->channel1 --->sink1
单source ---> channel----->sink2 source
----->sink3 ------>channel2---->sink2
Flume支持将事件流向一个或者多个目的地。这种模式将数据源复制到多个channel中，每一个channel都有相同的数据，sink能够选择传送的不一样的目的地。
③ 负载均衡 Flume支持使用将多个sink逻辑上分到一个sink组，flume将数据发送到不一样的sink，主要解决负载均衡和故障转移问题。
负载均衡 ：并排的三个channel都是轮询，好处是增大流量而且保证数据的安全；（一个挂了，三个不会都挂；缓冲区比较长，若是hdfs出现问题，两层的channel，多个flune的并联能够保证数据的安全且增大缓冲区）
④ Flume agent聚合 平常web应用一般分布在上百个服务器，大者甚至上千个、上万个服务器。产生的日志，处理起来也很是麻烦。用flume的这种组合方式能很好的解决这一问题，每台服务器部署一个flume采集日志，传送到一个集中收集日志的flume，再由此flume上传到hdfs、hive、hbase、jms等，进行日志分析。
安装
将apache-flume-1.7.0-bin.tar.gz上传到linux的/opt/software目录下 解压apache-flume-1.7.0-bin.tar.gz到/opt/module/目录下 [kris@hadoop101 software]$ tar -zxf apache-flume-1.7.0-bin.tar.gz -C /opt/module/ [kris@hadoop101 module]$ mv apache-flume-1.7.0-bin/ flume [kris@hadoop101 conf]$ mv flume-env.sh.template flume-env.sh [kris@hadoop101 conf]$ vim flume-env.sh export JAVA_HOME=/opt/module/jdk1.8.0_144
Flume异常处理
1）问题描述：若是启动消费Flume抛出以下异常
ERROR hdfs.HDFSEventSink: process failed
java.lang.OutOfMemoryError: GC overhead limit exceeded
2）解决方案步骤：
（1）在hadoop101服务器的/opt/module/flume/conf/flume-env.sh文件中增长以下配置
export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"
同步配置到hadoop10二、hadoop103服务器
[kris@hadoop101 conf]$ xsync flume-env.sh
1. 监控端口数据--netcat
监控端口数据:
端口(netcat)--->flume--->Sink(logger)到控制台
[kris@hadoop101 flume]$ mkdir job [kris@hadoop101 flume]$ cd job/ [kris@hadoop101 job]$ touch flume-netcat-logger.conf [kris@hadoop101 job]$ vim flume-netcat-logger.conf
# 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的输入源类型为netcat端口类型 a1.sources.r1.bind = localhost #表示a1的主机 a1.sources.r1.port = 44444 #表示a1的监听端口 # Describe the sink a1.sinks.k1.type = logger #表示a1的输出目的地是控制台logger类型 # Use a channel which buffers events in memory a1.channels.c1.type = memory #表示a1的channel类型是memory内存类型 a1.channels.c1.capacity = 1000 #表示a1的channel总容量是1000个event a1.channels.c1.transactionCapacity = 100 #表示a1的channel传输时收集到了100条event之后再去提交到事务 # Bind the source and sink to the channel a1.sources.r1.channels = c1 #表示r1和c1链接起来 a1.sinks.k1.channel = c1 #表示k1和c1链接起来
安装nc工具 [kris@hadoop101 software]$ sudo yum install -y nc 判断44444端口是否被占用 [kris@hadoop101 flume]$ sudo netstat -tunlp | grep 44444 先开启flume监听端口 [kris@hadoop101 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console --conf conf/表配置文件在conf/目录； --name a1是给agent起名a1； --conf-file job/...表本次读取配置文件是在job文件夹下的flume-netcat-logger.conf文件；
-D表flume运行时动态修改flume.root.logger参数属性值，并将控制台打印级别设置为INFO级别
[kris@hadoop101 ~]$ cd /opt/module/flume/ 向本机的44444端口发送内容 [kris@hadoop101 flume]$ nc localhost 44444 hello OK kris OK 在Flume监听页面观察接收数据状况 2019-02-20 10:01:41,151 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 68 65 6C 6C 6F hello } 2019-02-20 10:01:45,153 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 6B 72 69 73 kris } netstat -nltp [kris@hadoop101 ~]$ netstat -nltp #出现了监听这个端口号说明成功； tcp 0 0 ::ffff:127.0.0.1:44444 :::* LISTEN 4841/java
nc hadoop102 44444， flume不能接收到
netstat命令是一个监控TCP/IP网络的很是有用的工具，它能够显示路由表、实际的网络链接以及每个网络接口设备的状态信息。
-t或--tcp：显示TCP传输协议的连线情况；
-u或--udp：显示UDP传输协议的连线情况；
-n或--numeric：直接使用ip地址，而不经过域名服务器；
-l或--listening：显示监控中的服务器的Socket；
-p或--programs：显示正在使用Socket的程序识别码（PID）和程序名称；
2. 实时读取本地文件到HDFS
实时读取本地文件到HDFS:
hive.log(exec)--->flume--->Sink(HDFS)
取Linux系统中的文件，就得按照Linux命令的规则执行命令。因为Hive日志在Linux系统中因此读取文件的类型选择：exec即execute执行的意思。表示执行Linux命令来读取文件。
1．Flume要想将数据输出到HDFS，必须持有Hadoop相关jar包
将commons-configuration-1.6.jar、 hadoop-auth-2.7.2.jar、 hadoop-common-2.7.2.jar、 hadoop-hdfs-2.7.2.jar、 commons-io-2.4.jar、 htrace-core-3.1.0-incubating.jar 拷贝到/opt/module/flume/lib文件夹下
2．建立flume-file-hdfs.conf文件
[kris@hadoop101 job]$ vim flume-file-hdfs.conf
# Name the components on this agent a2.sources = r2 a2.sinks = k2 a2.channels = c2 # Describe/configure the source a2.sources.r2.type = exec a2.sources.r2.command = tail -F /opt/module/hive/logs/hive.log a2.sources.r2.shell = /bin/bash -c # Describe the sink a2.sinks.k2.type = hdfs a2.sinks.k2.hdfs.path = hdfs://hadoop102:9000/flume/%Y%m%d/%H #上传文件的前缀 a2.sinks.k2.hdfs.filePrefix = logs- #是否按照时间滚动文件夹 a2.sinks.k2.hdfs.round = true #多少时间单位建立一个新的文件夹 a2.sinks.k2.hdfs.roundValue = 1 #从新定义时间单位 a2.sinks.k2.hdfs.roundUnit = hour #是否使用本地时间戳 a2.sinks.k2.hdfs.useLocalTimeStamp = true #积攒多少个Event才flush到HDFS一次 a2.sinks.k2.hdfs.batchSize = 1000 #设置文件类型，可支持压缩 a2.sinks.k2.hdfs.fileType = DataStream #多久生成一个新的文件 a2.sinks.k2.hdfs.rollInterval = 60 #设置每一个文件的滚动大小 a2.sinks.k2.hdfs.rollSize = 134217700 #文件的滚动与Event数量无关 a2.sinks.k2.hdfs.rollCount = 0 # Use a channel which buffers events in memory a2.channels.c2.type = memory a2.channels.c2.capacity = 1000 a2.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a2.sources.r2.channels = c2 a2.sinks.k2.channel = c2
tail -F /opt/module/hive/logs/hive.log -F实时监控
[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a2 -f job/flume-file-hdfs.conf
开启Hadoop和Hive并操做Hive产生日志 sbin/start-dfs.sh sbin/start-yarn.sh bin/hive
在HDFS上查看文件。
3. 实时读取目录文件到HDFS
实时读取目录文件到HDFS:
目录dir(spooldir)--->flume--->Sink(HDFS)
[kris@hadoop101 job]$ vim flume-dir-hdfs.conf
a3.sources = r3 a3.sinks = k3 a3.channels = c3 # Describe/configure the source a3.sources.r3.type = spooldir #定义source类型为目录 a3.sources.r3.spoolDir = /opt/module/flume/upload #定义监控日志 a3.sources.r3.fileSuffix = .COMPLETED #定义文件上传完，后缀 a3.sources.r3.fileHeader = true #是否有文件头 #忽略全部以.tmp结尾的文件，不上传 a3.sources.r3.ignorePattern = ([^ ]*\.tmp) # Describe the sink a3.sinks.k3.type = hdfs a3.sinks.k3.hdfs.path = hdfs://hadoop101:9000/flume/upload/%Y%m%d/%H #上传文件的前缀 a3.sinks.k3.hdfs.filePrefix = upload- #是否按照时间滚动文件夹 a3.sinks.k3.hdfs.round = true #多少时间单位建立一个新的文件夹 a3.sinks.k3.hdfs.roundValue = 1 #从新定义时间单位 a3.sinks.k3.hdfs.roundUnit = hour #是否使用本地时间戳 a3.sinks.k3.hdfs.useLocalTimeStamp = true #积攒多少个Event才flush到HDFS一次 a3.sinks.k3.hdfs.batchSize = 100 #设置文件类型，可支持压缩 a3.sinks.k3.hdfs.fileType = DataStream #多久生成一个新的文件 a3.sinks.k3.hdfs.rollInterval = 60 #设置每一个文件的滚动大小大概是128M a3.sinks.k3.hdfs.rollSize = 134217700 #文件的滚动与Event数量无关 a3.sinks.k3.hdfs.rollCount = 0 # Use a channel which buffers events in memory a3.channels.c3.type = memory a3.channels.c3.capacity = 1000 a3.channels.c3.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r3.channels = c3 a3.sinks.k3.channel = c3
[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a3 -f job/flume-dir-hdfs.conf [kris@hadoop101 flume]$ mkdir upload [kris@hadoop101 flume]$ cd upload/ [kris@hadoop101 upload]$ touch kris.txt [kris@hadoop101 upload]$ touch kris.tmp [kris@hadoop101 upload]$ touch kris.log [kris@hadoop101 upload]$ ll ##建立文件，hdfs上就会生成/flume/upload/20190224/11/upload-155...的文件；不添加内容就是空的；vim kri.log.COMPLETED写入东西hdfs上仍是空的，它只是监控文件夹的建立； 总用量 0 -rw-rw-r--. 1 kris kris 0 2月 20 11:09 kris.log.COMPLETED -rw-rw-r--. 1 kris kris 0 2月 20 11:08 kris.tmp -rw-rw-r--. 1 kris kris 0 2月 20 11:08 kris.txt.COMPLETED [kris@hadoop101 flume]$ cp README.md upload/ [kris@hadoop101 flume]$ cp LICENSE upload/ [kris@hadoop101 upload]$ ll 总用量 32 -rw-rw-r--. 1 kris kris 0 2月 20 11:09 kris.log.COMPLETED -rw-rw-r--. 1 kris kris 0 2月 20 11:08 kris.tmp -rw-rw-r--. 1 kris kris 0 2月 20 11:08 kris.txt.COMPLETED -rw-r--r--. 1 kris kris 27625 2月 20 11:14 LICENSE.COMPLETED -rw-r--r--. 1 kris kris 2520 2月 20 11:13 README.md.COMPLETED 在upload中建立一个文件，就会在hdfs上建立一个文件；
也可在文件里边追加数据
4. 单数据源多出口（选择器）
单Source多Channel、Sink
单数据源多出口(选择器):单Source多Channel、Sink
hive.log(exec)---->flume1--Sink1(avro)-->flume2--->Sink(HDFS)
---Sink2(avro)-->flume3--->Sink(file roll本地目录文件data)
准备工做
在/opt/module/flume/job目录下建立group1文件夹 [kris@hadoop101 job]$ cd group1/ 在/opt/module/datas/目录下建立flume3文件夹 [kris@hadoop101 datas]$ mkdir flume3
1．建立flume-file-flume.conf
配置1个接收日志文件的source和两个channel、两个sink，分别输送给flume-flume-hdfs和flume-flume-dir。
[kris@hadoop101 group1]$ vim flume-file-flume.conf
# Name the components on this agent a1.sources = r1 a1.sinks = k1 k2 a1.channels = c1 c2 # 将数据流复制给全部channel a1.sources.r1.selector.type = replicating # Describe/configure the source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /opt/module/hive/logs/hive.log a1.sources.r1.shell = /bin/bash -c # Describe the sink # sink端的avro是一个数据发送者 a1.sinks.k1.type = avro a1.sinks.k1.hostname = hadoop101 a1.sinks.k1.port = 4141 a1.sinks.k2.type = avro a1.sinks.k2.hostname = hadoop101 a1.sinks.k2.port = 4142 # Describe the channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 a1.channels.c2.type = memory a1.channels.c2.capacity = 1000 a1.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 c2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c2
Avro是由Hadoop创始人Doug Cutting建立的一种语言无关的数据序列化和RPC框架。
注：RPC（Remote Procedure Call）—远程过程调用，它是一种经过网络从远程计算机程序上请求服务，而不须要了解底层网络技术的协议。
[kris@hadoop101 group1]$ vim flume-flume-hdfs.conf
# Name the components on this agent a2.sources = r1 a2.sinks = k1 a2.channels = c1 # Describe/configure the source # source端的avro是一个数据接收服务 a2.sources.r1.type = avro a2.sources.r1.bind = hadoop101 a2.sources.r1.port = 4141 # Describe the sink a2.sinks.k1.type = hdfs a2.sinks.k1.hdfs.path = hdfs://hadoop101:9000/flume2/%Y%m%d/%H #上传文件的前缀 a2.sinks.k1.hdfs.filePrefix = flume2- #是否按照时间滚动文件夹 a2.sinks.k1.hdfs.round = true #多少时间单位建立一个新的文件夹 a2.sinks.k1.hdfs.roundValue = 1 #从新定义时间单位 a2.sinks.k1.hdfs.roundUnit = hour #是否使用本地时间戳 a2.sinks.k1.hdfs.useLocalTimeStamp = true #积攒多少个Event才flush到HDFS一次 a2.sinks.k1.hdfs.batchSize = 100 #设置文件类型，可支持压缩 a2.sinks.k1.hdfs.fileType = DataStream #多久生成一个新的文件 a2.sinks.k1.hdfs.rollInterval = 600 #设置每一个文件的滚动大小大概是128M a2.sinks.k1.hdfs.rollSize = 134217700 #文件的滚动与Event数量无关 a2.sinks.k1.hdfs.rollCount = 0 # Describe the channel a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a2.sources.r1.channels = c1 a2.sinks.k1.channel = c1
[kris@hadoop101 group1]$ vim flume-flume-dir.conf
# Name the components on this agent a3.sources = r1 a3.sinks = k1 a3.channels = c2 # Describe/configure the source a3.sources.r1.type = avro a3.sources.r1.bind = hadoop101 a3.sources.r1.port = 4142 # Describe the sink a3.sinks.k1.type = file_roll a3.sinks.k1.sink.directory = /opt/module/data/flume3 # Describe the channel a3.channels.c2.type = memory a3.channels.c2.capacity = 1000 a3.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r1.channels = c2 a3.sinks.k1.channel = c2
执行配置文件 分别开启对应配置文件：flume-flume-dir，flume-flume-hdfs，flume-file-flume。//从sink端往source端开启 [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a3 -f job/group1/flume-flume-dir.conf [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a2 -f job/group1/flume-flume-hdfs.conf [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/group1/flume-file-flume.conf 启动Hadoop和Hive start-dfs.sh start-yarn.sh bin/hive
检查HDFS上数据
检查/opt/module/datas/flume3目录中数据
[kris@hadoop101 ~]$ cd /opt/module/datas/flume3/ [kris@hadoop101 flume3]$ ll 总用量 4 -rw-rw-r--. 1 kris kris 0 2月 20 11:49 1550634573721-1 -rw-rw-r--. 1 kris kris 0 2月 20 11:54 1550634573721-10 -rw-rw-r--. 1 kris kris 0 2月 20 11:54 1550634573721-11 -rw-rw-r--. 1 kris kris 0 2月 20 11:50 1550634573721-2 -rw-rw-r--. 1 kris kris 0 2月 20 11:50 1550634573721-3 -rw-rw-r--. 1 kris kris 0 2月 20 11:51 1550634573721-4 -rw-rw-r--. 1 kris kris 0 2月 20 11:51 1550634573721-5 -rw-rw-r--. 1 kris kris 0 2月 20 11:52 1550634573721-6 -rw-rw-r--. 1 kris kris 0 2月 20 11:52 1550634573721-7 -rw-rw-r--. 1 kris kris 0 2月 20 11:53 1550634573721-8 -rw-rw-r--. 1 kris kris 1738 2月 20 11:53 1550634573721-9 [kris@hadoop101 flume3]$ cat 1550634573721-9 2019-02-20 11:00:42,459 INFO [main]: metastore.hivemetastoressimpl (HiveMetaStoreFsImpl.java:deleteDir(53)) - Deleted the diretory hdfs://hadoop101:9000/user/hive/warehouse/student22 2019-02-20 11:00:42,460 INFO [main]: log.PerfLogger (PerfLogger.java:PerfLogEnd(148)) - </PERFLOG method=runTasks start=1550631641861 end=1550631642460 duration=599 from=org.apache.hadoop.hive.ql.Driver> 2019-02-20 11:00:42,461 INFO [main]: log.PerfLogger (PerfLogger.java:PerfLogEnd(148)) - </PERFLOG method=Driver.execute start=1550631641860 end=1550631642461 duration=601 from=org.apache.hadoop.hive.ql.Driver> 2019-02-20 11:00:42,461 INFO [main]: ql.Driver (SessionState.java:printInfo(951)) - OK 2019-02-20 11:00:42,461 INFO [main]: log.PerfLogger (PerfLogger.java:PerfLogBegin(121)) - <PERFLOG method=releaseLocks from=org.apache.hadoop.hive.ql.Driver> 2019-02-20 11:00:42,461 INFO [main]: log.PerfLogger (PerfLogger.java:PerfLogEnd(148)) - </PERFLOG method=releaseLocks start=1550631642461 end=1550631642461 duration=0 from=org.apache.hadoop.hive.ql.Driver> 2019-02-20 11:00:42,461 INFO [main]: log.PerfLogger (PerfLogger.java:PerfLogEnd(148)) - </PERFLOG method=Driver.run start=1550631641638 end=1550631642461 duration=823 from=org.apache.hadoop.hive.ql.Driver> 2019-02-20 11:00:42,461 INFO [main]: CliDriver (SessionState.java:printInfo(951)) - Time taken: 0.824 seconds 2019-02-20 11:00:42,461 INFO [main]: log.PerfLogger (PerfLogger.java:PerfLogBegin(121)) - <PERFLOG method=releaseLocks from=org.apache.hadoop.hive.ql.Driver> 2019-02-20 11:00:42,462 INFO [main]: log.PerfLogger (PerfLogger.java:PerfLogEnd(148)) - </PERFLOG method=releaseLocks start=1550631642461 end=1550631642462 duration=1 from=org.apache.hadoop.hive.ql.Driver>
5. 单数据源多出口案例(Sink组)
单Source、Channel多Sink(负载均衡)
Flume 的负载均衡和故障转移
目的是为了提升整个系统的容错能力和稳定性。简单配置就能够轻松实现，首先须要设置 Sink 组，同一个 Sink 组内有多个子 Sink，不一样 Sink 之间能够配置成负载均衡或者故障转移。
单数据源多出口(Sink组): flum1-load_balance
端口(netcat)--->flume1---Sink1(avro)-->flume2---Sink(Logger控制台)
---Sink2(avro)-->flume3---Sink(Logger控制台)
flume1配置了数据均衡的输出到各个sink端：见下
[kris@hadoop101 group2]$ cat flume-netcat-flume.conf
# Name the components on this agent a1.sources = r1 a1.channels = c1 a1.sinkgroups = g1 a1.sinks = k1 k2 # Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 a1.sinkgroups.g1.processor.type = load_balance a1.sinkgroups.g1.processor.backoff = true a1.sinkgroups.g1.processor.selector = round_robin a1.sinkgroups.g1.processor.selector.maxTimeOut=10000 # Describe the sink a1.sinks.k1.type = avro a1.sinks.k1.hostname = hadoop101 a1.sinks.k1.port = 4141 a1.sinks.k2.type = avro a1.sinks.k2.hostname = hadoop101 a1.sinks.k2.port = 4142 # Describe the channel 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.sinkgroups.g1.sinks = k1 k2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c1
[kris@hadoop101 group2]$ cat flume-flume-console1.conf
# Name the components on this agent a2.sources = r1 a2.sinks = k1 a2.channels = c1 # Describe/configure the source a2.sources.r1.type = avro a2.sources.r1.bind = hadoop101 a2.sources.r1.port = 4141 # Describe the sink a2.sinks.k1.type = logger # Describe the channel a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a2.sources.r1.channels = c1 a2.sinks.k1.channel = c1
[kris@hadoop101 group2]$ cat flume-flume-console2.conf
# Name the components on this agent a3.sources = r1 a3.sinks = k1 a3.channels = c2 # Describe/configure the source a3.sources.r1.type = avro a3.sources.r1.bind = hadoop101 a3.sources.r1.port = 4142 # Describe the sink a3.sinks.k1.type = logger # Describe the channel a3.channels.c2.type = memory a3.channels.c2.capacity = 1000 a3.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r1.channels = c2 a3.sinks.k1.channel = c2
[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a3 -f job/group2/flume-flume-console2.conf -Dflume.root.logger=INFO,console [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a2 -f job/group2/flume-flume-console1.conf -Dflume.root.logger.INFO,console [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/group2/flume-netcat-flume.conf
[kris@hadoop101 group2]$ nc localhost 44444 1 OK 1 OK 2 OK 3 OK 4 oggerSink.java:95)] Event: { headers:{} body: 31 1 } 2019-02-20 15:26:37,828 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 31 1 } 2019-02-20 15:26:37,828 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 32 2 } 2019-02-20 15:26:37,829 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 33 3 } 2019-02-20 15:26:37,829 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 34 4 } 2019-02-20 15:26:37,830 (SinkRunne 2019-02-20 15:27:06,706 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 61 a } 2019-02-20 15:27:06,706 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 62 b } 2019-02-20 15:27:06,707
6. 多数据源汇总
多Source汇总数据到单Flume
7. 多数据源汇总:
group.log(exec)--->flume1--Sink(avro；hadoop103--4141)-->flume3---Sink(Logger控制台)
端口(netcat)-->flume2--Sink(avro；hadoop103-4141)-->flume3---Sink(Logger控制台)
分发Flume
[kris@hadoop101 module]$ xsync flume 在hadoop10一、hadoop102以及hadoop103的/opt/module/flume/job目录下建立一个group3文件夹。 [kris@hadoop101 job]$ mkdir group3 [kris@hadoop102 job]$ mkdir group3 [kris@hadoop103 job]$ mkdir group3
1．建立flume1-logger-flume.conf
配置Source用于监控hive.log文件，配置Sink输出数据到下一级Flume。
在hadoop102上建立配置文件并打开
[kris@hadoop102 group3]$ vim flume1-logger-flume.conf
# Name the components on this agent # Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /opt/module/group.log a1.sources.r1.shell = /bin/bash -c # Describe the sink a1.sinks.k1.type = avro a1.sinks.k1.hostname = hadoop103 a1.sinks.k1.port = 4141 # Describe the channel 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
2．建立flume2-netcat-flume.conf
配置Source监控端口44444数据流，配置Sink数据到下一级Flume：
在hadoop101上建立配置文件并打开
[kris@hadoop101 group3]$ vim flume2-netcat-flume.conf
# Name the components on this agent # Name the components on this agent a2.sources = r1 a2.sinks = k1 a2.channels = c1 # Describe/configure the source a2.sources.r1.type = netcat a2.sources.r1.bind = hadoop101 a2.sources.r1.port = 44444 # Describe the sink a2.sinks.k1.type = avro a2.sinks.k1.hostname = hadoop103 a2.sinks.k1.port = 4141 # Use a channel which buffers events in memory a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a2.sources.r1.channels = c1 a2.sinks.k1.channel = c1
3．建立flume3-flume-logger.conf
配置source用于接收flume1与flume2发送过来的数据流，最终合并后sink到控制台。
在hadoop103上建立配置文件并打开；由于前面两个avro都是hadoop103: 4141，它们的ip和端口是同样的，因此只需配置一个avro便可
[kris@hadoop103 group3]$ vim flume3-flume-logger.conf
# Name the components on this agent a3.sources = r1 a3.sinks = k1 a3.channels = c1 # Describe/configure the source a3.sources.r1.type = avro a3.sources.r1.bind = hadoop103 a3.sources.r1.port = 4141 # Describe the sink # Describe the sink a3.sinks.k1.type = logger # Describe the channel a3.channels.c1.type = memory a3.channels.c1.capacity = 1000 a3.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r1.channels = c1 a3.sinks.k1.channel = c1
4．执行配置文件
分别开启对应配置文件：flume3-flume-logger.conf，flume2-netcat-flume.conf，flume1-logger-flume.conf。
[kris@hadoop103 flume]$ bin/flume-ng agent -c conf/ -n a3 -f job/group3/flume3-flume-logger.conf -Dflume.root.logger=INFO,console [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a2 -f job/group3/flume2-netcat-flume.conf [kris@hadoop102 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/group3/flume1-logger-flume.conf
在hadoop102上向/opt/module目录下的group.log追加内容 [kris@hadoop102 module]$ echo "Hello World" > group.log [kris@hadoop102 module]$ ll 总用量 24 drwxrwxr-x. 10 kris kris 4096 2月 20 11:07 flume -rw-rw-r--. 1 kris kris 12 2月 20 16:13 group.log 在hadoop101上向44444端口发送数据 [kris@hadoop101 flume]$ nc hadoop101 44444 1 OK 2 OK 3 OK 4 检查hadoop103上数据 2019-02-20 16:13:20,748 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 48 65 6C 6C 6F 20 57 6F 72 6C 64 Hello World } 2019-02-20 16:14:46,774 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 31 1 } 2019-02-20 16:14:46,775 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 32 2 }
8. 练习
案例需求：
1）flume-1监控hive.log日志，flume-1的数据传送给flume-2，flume-2将数据追加到本地文件，同时将数据传输到flume-3。
2）flume-4监控本地另外一个本身建立的文件any.txt，并将数据传送给flume-3。
3）flume-3将汇总数据写入到HDFS。
请先画出结构图，再开始编写任务脚本。
hive.log(exec)--->flume-1 ---Sink1(avro;hadoop101:4141) --> flume-2--Sink1(logger本地文件) --Sink2(avro;hadoop101:4142) --> flume-3--Sink(HDFS) 本地any.txt(exec)--->flume-4--Sink(avro;hadoop101:4142)-->flume-3到HDFS 启动三、二、一、4
flume-1:
vim flume1-file-flume.conf
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /opt/module/hive/logs/hive.log a1.sources.r1.shell = /bin/bash -c # Describe the sink # sink端的avro是一个数据发送者 a1.sinks.k1.type = avro a1.sinks.k1.hostname = hadoop101 a1.sinks.k1.port = 4141 # Describe the channel 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
flume2:
vim flume2-flume-dir.conf
# Name the components on this agent a2.sources = r1 a2.sinks = k1 k2 a2.channels = c1 c2 # 将数据流复制给全部channel a2.sources.r1.selector.type = replicating # Describe/configure the source a2.sources.r1.type = avro a2.sources.r1.bind = hadoop101 a2.sources.r1.port = 4141 # Describe the sink a2.sinks.k1.type = logger a2.sinks.k2.type = avro a2.sinks.k2.hostname = hadoop101 a2.sinks.k2.port = 4142 # Describe the channel a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 # Describe the channel a2.channels.c2.type = memory a2.channels.c2.capacity = 1000 a2.channels.c2.transactionCapacity = 100 # Bind the source and sink to the channel a2.sources.r1.channels =c1 c2 a2.sinks.k1.channel = c1 a2.sinks.k2.channel = c2
flume3:
vim flume3-flume-hdfs.conf
# Name the components on this agent a3.sources = r1 a3.sinks = k1 a3.channels = c1 # Describe/configure the source # source端的avro是一个数据接收服务 a3.sources.r1.type = avro a3.sources.r1.bind = hadoop101 a3.sources.r1.port = 4142 # Describe the sink a3.sinks.k1.type = hdfs a3.sinks.k1.hdfs.path = hdfs://hadoop101:9000/flume3/%Y%m%d/%H #上传文件的前缀 a3.sinks.k1.hdfs.filePrefix = flume3- #是否按照时间滚动文件夹 a3.sinks.k1.hdfs.round = true #多少时间单位建立一个新的文件夹 a3.sinks.k1.hdfs.roundValue = 1 #从新定义时间单位 a3.sinks.k1.hdfs.roundUnit = hour #是否使用本地时间戳 a3.sinks.k1.hdfs.useLocalTimeStamp = true #积攒多少个Event才flush到HDFS一次 a3.sinks.k1.hdfs.batchSize = 100 #设置文件类型，可支持压缩 a3.sinks.k1.hdfs.fileType = DataStream #多久生成一个新的文件 a3.sinks.k1.hdfs.rollInterval = 600 #设置每一个文件的滚动大小大概是128M a3.sinks.k1.hdfs.rollSize = 134217700 #文件的滚动与Event数量无关 a3.sinks.k1.hdfs.rollCount = 0 # Describe the channel a3.channels.c1.type = memory a3.channels.c1.capacity = 1000 a3.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a3.sources.r1.channels = c1 a3.sinks.k1.channel = c1
flume4:
vim flume4-file-flume.conf
# Name the components on this agent a4.sources = r1 a4.sinks = k1 a4.channels = c1 # Describe/configure the source a4.sources.r1.type = exec a4.sources.r1.command = tail -F /opt/module/datas/any.txt a4.sources.r1.shell = /bin/bash -c # Describe the sink # sink端的avro是一个数据发送者 a4.sinks.k1.type = avro a4.sinks.k1.hostname = hadoop101 a4.sinks.k1.port = 4142 # Describe the channel a4.channels.c1.type = memory a4.channels.c1.capacity = 1000 a4.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a4.sources.r1.channels = c1 a4.sinks.k1.channel = c1
启动
[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a3 -f job/group4/flume3-flume-hdfs.conf [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a2 -f job/group4/flume2-flume-dir.conf -Dflume.root.logger=INFO,console [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/group4/flume1-file-flume.conf [kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a4 -f job/group4/flume4-file-flume.conf 数据来源：flume1；hive.log和flume4| any.txt文件 [kris@hadoop101 datas]$ cat any.txt ##文件发生变化hdfs上会实时更新 1 2 3 4 5 《疑犯追踪》 悬疑,动做,科幻,剧情 《Lie to me》 悬疑,警匪,动做,心理,剧情 《战狼2》 战争,动做,灾难 II Love You [kris@hadoop101 datas]$ pwd /opt/module/datas
9. 自定义Source
Source是负责接收数据到Flume Agent的组件。Source组件能够处理各类类型、各类格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy。官方提供的source类型已经不少，可是有时候并不能知足实际开发当中的需求，此时咱们就须要根据实际需求自定义某些source。
官方也提供了自定义source的接口：
https://flume.apache.org/FlumeDeveloperGuide.html#source根据官方说明自定义MySource须要继承AbstractSource类并实现Configurable和PollableSource接口。
实现相应方法：
getBackOffSleepIncrement()//暂不用
getMaxBackOffSleepInterval()//暂不用
configure(Context context)//初始化context（读取配置文件内容）
process()//获取数据封装成event并写入channel，这个方法将被循环调用。
使用场景：读取MySQL数据或者其余文件系统。
需求：使用flume接收数据，并给每条数据添加前缀，输出到控制台。前缀可从flume配置文件中配置。
import org.apache.flume.Context; import org.apache.flume.EventDeliveryException; import org.apache.flume.PollableSource; import org.apache.flume.conf.Configurable; import org.apache.flume.event.SimpleEvent; import org.apache.flume.source.AbstractSource; import java.util.HashMap; import java.util.Map; public class MySource extends AbstractSource implements Configurable, PollableSource { private String prefix; private Long delay; /** * 数据处理方法,被flume循环调用 * @return 数据读取状态 * @throws EventDeliveryException 咱们异常就回滚 */ public Status process() throws EventDeliveryException { Status status = null; //Status是个enum类型，成功或失败 //建立事件 SimpleEvent event = new SimpleEvent(); //Event由可选的header和载有数据的一个byte array 构成 Map<String, String> headerMap = new HashMap<String, String>(); //Header是容纳了key-value字符串对的HashMap。 for (int i = 0; i < 5; i++){ try { event.setHeaders(headerMap); //封装事件 event.setBody((prefix + "LLL" + i).getBytes()); getChannelProcessor().processEvent(event);//将事件写入channel status = Status.READY; Thread.sleep(delay); } catch (InterruptedException e) { e.printStackTrace(); return Status.BACKOFF; } } return status; } public long getBackOffSleepIncrement() { return 0; } public long getMaxBackOffSleepInterval() { return 0; } /** * 配置自定义的Source * @param context */ public void configure(Context context) { prefix = context.getString("prefix", "Hello"); delay = context.getLong("delay", 1000L); } }
测试
1）打包
将写好的代码打包，并放到flume的lib目录（/opt/module/flume）下。
2）配置文件
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = com.atguigu.source.MySource a1.sources.r1.delay = 1000 #a1.sources.r1.field = HelloWorld # 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
[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/mysource-flume-logger.conf -Dflume.root.logger=INFO,console
10. 自定义Sink
Sink不断地轮询Channel中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另外一个Flume Agent。
Sink是彻底事务性的。在从Channel批量删除数据以前，每一个Sink用Channel启动一个事务。批量事件一旦成功写出到存储系统或下一个Flume Agent，Sink就利用Channel提交事务。事务一旦被提交，该Channel从本身的内部缓冲区删除事件。
Sink组件目的地包括hdfs、logger、avro、thrift、ipc、file、null、HBase、solr、自定义。官方提供的Sink类型已经不少，可是有时候并不能知足实际开发当中的需求，此时咱们就须要根据实际需求自定义某些Sink。
官方也提供了自定义source的接口：
https://flume.apache.org/FlumeDeveloperGuide.html#sink根据官方说明自定义MySink须要继承AbstractSink类并实现Configurable接口。
实现相应方法：
configure(Context context)//初始化context（读取配置文件内容）
process()//从Channel读取获取数据（event），这个方法将被循环调用。
使用场景：读取Channel数据写入MySQL或者其余文件系统。
需求：使用flume接收数据，并在Sink端给每条数据添加前缀和后缀，输出到控制台。先后缀可在flume任务配置文件中配置。
import org.apache.flume.*; import org.apache.flume.conf.Configurable; import org.apache.flume.sink.AbstractSink; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class MySink extends AbstractSink implements Configurable { private static final Logger LOGGER = LoggerFactory.getLogger(AbstractSink.class); //建立logger对象 private String prefix; private String suffix; /** * Sink从channel中拉取数据并处理 * @return * @throws EventDeliveryException */ public Status process() throws EventDeliveryException { Status status = null; //声明返回值状态 Event event;//声明事件 Channel channel = getChannel();//获取当前sink绑定的channel Transaction transaction = channel.getTransaction();//获取事务 transaction.begin(); //开启事务 try { while ((event = channel.take()) == null) { Thread.sleep(500); } LOGGER.info(prefix + new String(event.getBody()) + suffix); status = Status.READY; transaction.commit(); //事务提交 } catch (Exception e) { e.printStackTrace(); status = Status.BACKOFF; transaction.rollback(); //事务回滚 } finally { transaction.close(); //关闭事务 } return status; } /** * 设置Sink * @param context 上下文环境 */ public void configure(Context context) { prefix = context.getString("prefix", "Hello"); suffix = context.getString("suffix", "kris"); } }
测试
1）打包
将写好的代码打包，并放到flume的lib目录（/opt/module/flume）下。
2）配置文件
# 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 = com.atguigu.source.MySink #a1.sinks.k1.prefix = kris: a1.sinks.k1.suffix = :kris # 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
[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/mysource-flume-netcat.conf -Dflume.root.logger=INFO,console [kris@hadoop101 job]$ nc localhost 44444 1 OK 2 2019-02-24 16:27:25,078 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - com.atguigu.source.MySink.process(MySink.java:32)] kris:1:kris 2019-02-24 16:27:25,777 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - com.atguigu.source.MySink.process(MySink.java:32)] kris:2:kris
11. Flume监控之Ganglia
Ganglia的安装与部署
安装ganglia 、httpd服务与php、其余依赖
sudo rpm -Uvh http://dl.fedoraproject.org/pub/epel/6/x86_64/epel-release-6-8.noarch.rpm sudo yum -y install httpd php rrdtool perl-rrdtool rrdtool-devel apr-devel ganglia-gmetad ganglia-web ganglia-gmond
Ganglia由gmond、gmetad和gweb三部分组成。
gmond（Ganglia Monitoring Daemon）是一种轻量级服务，安装在每台须要收集指标数据的节点主机上。使用gmond，你能够很容易收集不少系统指标数据，如CPU、内存、磁盘、网络和活跃进程的数据等。
gmetad（Ganglia Meta Daemon）整合全部信息，并将其以RRD格式存储至磁盘的服务。
gweb（Ganglia Web）Ganglia可视化工具，gweb是一种利用浏览器显示gmetad所存储数据的PHP前端。在Web界面中以图表方式展示集群的运行状态下收集的多种不一样指标数据。
配置
1）修改配置文件/etc/httpd/conf.d/ganglia.conf [kris@hadoop101 flume]$ sudo vim /etc/httpd/conf.d/ganglia.conf 2）修改成红颜色的配置： # Ganglia monitoring system php web frontend Alias /ganglia /usr/share/ganglia <Location /ganglia> Order deny,allow #Deny from all Allow from all # Allow from 127.0.0.1 # Allow from ::1 # Allow from .example.com </Location> 3) 修改配置文件/etc/ganglia/gmetad.conf [kris@hadoop101 flume]$ sudo vim /etc/ganglia/gmetad.conf 修改成： data_source "hadoop101" 192.168.1.101 3) 修改配置文件/etc/ganglia/gmond.conf [kris@hadoop101 flume]$ sudo vim /etc/ganglia/gmond.conf 修改成： cluster { name = "hadoop101" owner = "unspecified" latlong = "unspecified" url = "unspecified" } udp_send_channel { #bind_hostname = yes # Highly recommended, soon to be default. # This option tells gmond to use a source address # that resolves to the machine's hostname. Without # this, the metrics may appear to come from any # interface and the DNS names associated with # those IPs will be used to create the RRDs. # mcast_join = 239.2.11.71 host = 192.168.1.101 port = 8649 ttl = 1 } udp_recv_channel { # mcast_join = 239.2.11.71 port = 8649 bind = 192.168.1.101 retry_bind = true # Size of the UDP buffer. If you are handling lots of metrics you really # should bump it up to e.g. 10MB or even higher. # buffer = 10485760 } 4) 修改配置文件/etc/selinux/config [kris@hadoop101 flume]$ sudo vim /etc/selinux/config 修改成： # This file controls the state of SELinux on the system. # SELINUX= can take one of these three values: # enforcing - SELinux security policy is enforced. # permissive - SELinux prints warnings instead of enforcing. # disabled - No SELinux policy is loaded. SELINUX=disabled # SELINUXTYPE= can take one of these two values: # targeted - Targeted processes are protected, # mls - Multi Level Security protection. SELINUXTYPE=targeted
尖叫提示：selinux本次生效关闭必须重启，若是此时不想重启，能够临时生效之： [kris@hadoop101 flume]$ sudo setenforce 0
1.启动
1) 启动ganglia [kris@hadoop101 flume]$ sudo service httpd start [kris@hadoop101 flume]$ sudo service gmetad start [kris@hadoop101 flume]$ sudo service gmond start 2) 打开网页浏览ganglia页面 http://192.168.1.101/ganglia 尖叫提示：若是完成以上操做依然出现权限不足错误，请修改/var/lib/ganglia目录的权限： [kris@hadoop101 flume]$ sudo chmod -R 777 /var/lib/ganglia
2 操做Flume测试监控 1) 修改/opt/module/flume/conf目录下的flume-env.sh配置： JAVA_OPTS="-Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.1.101:8649 -Xms100m -Xmx200m" 2) 启动Flume任务 [kris@hadoop101 flume]$ bin/flume-ng agent \ --conf conf/ \ --name a1 \ --conf-file job/flume-netcat-logger.conf \ -Dflume.root.logger==INFO,console \ -Dflume.monitoring.type=ganglia \ -Dflume.monitoring.hosts=192.168.1.101:8649
简写以下：
[kris@hadoop101 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=192.168.1.101:8649
3) 发送数据观察ganglia监测图 [kris@hadoop101 flume]$ nc localhost 44444
flume.CHANNEL.c1.EventPutSuccessCount flume发送的单例叫event，put叫成功接收的数据，就是往channel里边put的数据
flume.CHANNEL.c1.EventTakeSuccessCount 这个是take的数据，更日志数据作对比看有没有丢数据
flume.CHANNEL.c1.ChannelFillPercentage 这个数只要不满，就不会丢数据，若是1.0表示所有填满了；
- 1. Flume之flume-ng
- 2. Flume>Flume介绍
- 3. Flume（一）—— Flume介绍
- 4. 【Flume 1.6.0】5、Flume Interceptors
- 5. flume-1 flume概述
- 6. flume flume source组件
- 7. 【Flume一】Flume入门
- 8. Flume
- 9. flume
- 10. 【Flume】
- 更多相关文章...
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. Flume之flume-ng
- 2. Flume>Flume介绍
- 3. Flume（一）—— Flume介绍
- 4. 【Flume 1.6.0】5、Flume Interceptors
- 5. flume-1 flume概述
- 6. flume flume source组件
- 7. 【Flume一】Flume入门
- 8. Flume
- 9. flume
- 10. 【Flume】