大数据之Flume
本文共 46674 字，大约阅读时间需要 155 分钟。
Flume定义
Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚和传输的系统。Flume基于流式架构，灵活简单。
. Flume最主要的作用就是，实时读取服务器本地磁盘的数据，将数据写入到HDFS。Flume的优点
1.可以和任意存储进程集成。
2.输入的的数据速率大于写入目的存储的速率，flume会进行缓冲，减小hdfs的压力。 3.flume中的事务基于channel，使用了两个事务模型（sender + receiver），确保消息被可靠发送。 Flume使用两个独立的事务分别负责从source到channel，以及从channel到sink的事件传递。一旦事务中所有的数据全部成功提交到channel，那么source才认为该数据读取完成。同理，只有成功被sink写出去的数据，才会从channel中移除。Flume组成架构
Agent
Agent是一个JVM进程，它以事件的形式将数据从源头送至目的。
Agent主要有3个部分组成，Source、Channel、Sink。Source
Source是负责接收数据到Flume Agent的组件。Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy。
Channel
Channel是位于Source和Sink之间的缓冲区。因此，Channel允许Source和Sink运作在不同的速率上。Channel是线程安全的，可以同时处理几个Source的写入操作和几个Sink的读取操作。
Flume自带两种Channel：Memory Channel和File Channel。 Memory Channel是内存中的队列。Memory Channel在不需要关心数据丢失的情景下适用。如果需要关心数据丢失，那么Memory Channel就不应该使用，因为程序死亡、机器宕机或者重启都会导致数据丢失。 File Channel将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据。Sink
Sink不断地轮询Channel中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个Flume Agent。
Sink是完全事务性的。在从Channel批量删除数据之前，每个Sink用Channel启动一个事务。批量事件一旦成功写出到存储系统或下一个Flume Agent，Sink就利用Channel提交事务。事务一旦被提交，该Channel从自己的内部缓冲区删除事件。 Sink组件目的地包括hdfs、logger、avro、thrift、ipc、file、null、HBase、solr、自定义。Event
传输单元，Flume数据传输的基本单元，以事件的形式将数据从源头送至目的地。 Event由可选的header和载有数据的一个byte array 构成。Header是容纳了key-value字符串对的HashMap。
Flume Agent内部原理
企业开发案例
监控端口数据官方案例
1）案例需求：
首先启动Flume任务，监控本机44444端口，服务端； 然后通过netcat工具向本机44444端口发送消息，客户端； 最后Flume将监听的数据实时显示在控制台。 2）需求分析：
3）实现步骤： 1．安装netcat工具 [jinghang@hadoop102 software]$ sudo yum install -y nc
2．判断44444端口是否被占用
[jinghang@hadoop102 flume-telnet]$ sudo netstat -tunlp | grep 44444
功能描述：netstat命令是一个监控TCP/IP网络的非常有用的工具，它可以显示路由表、实际的网络连接以及每一个网络接口设备的状态信息。
基本语法：netstat [选项] 选项参数： -t或–tcp：显示TCP传输协议的连线状况； -u或–udp：显示UDP传输协议的连线状况； -n或–numeric：直接使用ip地址，而不通过域名服务器； -l或–listening：显示监控中的服务器的Socket； -p或–programs：显示正在使用Socket的程序识别码（PID）和程序名称； 3．创建Flume Agent配置文件flume-netcat-logger.conf 在flume目录下创建job文件夹并进入job文件夹。[jinghang@hadoop102 flume]$ mkdir job[jinghang@hadoop102 flume]$ cd job/
在job文件夹下创建Flume Agent配置文件flume-netcat-logger.conf。
[jinghang@hadoop102 job]$ touch flume-netcat-logger.conf
在flume-netcat-logger.conf文件中添加如下内容。
[jinghang@hadoop102 job]$ vim flume-netcat-logger.conf
添加内容如下：
# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the sourcea1.sources.r1.type = netcata1.sources.r1.bind = localhosta1.sources.r1.port = 44444# Describe the sinka1.sinks.k1.type = logger# Use a channel which buffers events in memorya1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1注：配置文件来源于官方手册http://flume.apache.org/FlumeUserGuide.html
- 先开启flume监听端口 第一种写法：
[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console
第二种写法：
[jinghang@hadoop102 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console
参数说明：
–conf conf/ ：表示配置文件存储在conf/目录 –name a1 ：表示给agent起名为a1 –conf-file job/flume-netcat.conf ：flume本次启动读取的配置文件是在job文件夹下的flume-telnet.conf文件。 -Dflume.root.logger==INFO,console ：-D表示flume运行时动态修改flume.root.logger参数属性值，并将控制台日志打印级别设置为INFO级别。日志级别包括:log、info、warn、error。 5．使用netcat工具向本机的44444端口发送内容[jinghang@hadoop102 ~]$ nc localhost 44444hello jinghang
6．在Flume监听页面观察接收数据情况
实时读取本地文件到HDFS案例
1）案例需求：实时监控Hive日志，并上传到HDFS中
2）需求分析：
3）实现步骤： 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文件 创建文件 [jinghang@hadoop102 job]$ touch flume-file-hdfs.conf
注：要想读取Linux系统中的文件，就得按照Linux命令的规则执行命令。由于Hive日志在Linux系统中所以读取文件的类型选择：exec即execute执行的意思。表示执行Linux命令来读取文件。
[jinghang@hadoop102 job]$ vim flume-file-hdfs.conf
添加如下内容
# Name the components on this agenta2.sources = r2a2.sinks = k2a2.channels = c2# Describe/configure the sourcea2.sources.r2.type = execa2.sources.r2.command = tail -F /opt/module/hive/logs/hive.loga2.sources.r2.shell = /bin/bash -c# Describe the sinka2.sinks.k2.type = hdfsa2.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 memorya2.channels.c2.type = memorya2.channels.c2.capacity = 1000a2.channels.c2.transactionCapacity = 100# Bind the source and sink to the channela2.sources.r2.channels = c2a2.sinks.k2.channel = c2
注意：
- 对于所有与时间相关的转义序列，Event Header中必须存在以 “timestamp”的key（除非hdfs.useLocalTimeStamp设置为true，此方法会使用TimestampInterceptor自动添加timestamp）。 a3.sinks.k3.hdfs.useLocalTimeStamp = true
3．执行监控配置
[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a2 --conf-file job/flume-file-hdfs.conf
4．开启Hadoop和Hive并操作Hive产生日志
[jinghang@hadoop102 hadoop-2.7.2]$ sbin/start-dfs.sh[jinghang@hadoop103 hadoop-2.7.2]$ sbin/start-yarn.sh[jinghang@hadoop102 hive]$ bin/hive
5．在HDFS上查看文件。
实时读取目录文件到HDFS案例
1）案例需求：使用Flume监听整个目录的文件
2）需求分析：
3）实现步骤：
1．创建配置文件flume-dir-hdfs.conf 创建一个文件[jinghang@hadoop102 job]$ touch flume-dir-hdfs.conf
打开文件
[jinghang@hadoop102 job]$ vim flume-dir-hdfs.conf
添加如下内容
a3.sources = r3a3.sinks = k3a3.channels = c3# Describe/configure the sourcea3.sources.r3.type = spooldira3.sources.r3.spoolDir = /opt/module/flume/uploada3.sources.r3.fileSuffix = .COMPLETEDa3.sources.r3.fileHeader = true#忽略所有以.tmp结尾的文件，不上传a3.sources.r3.ignorePattern = ([^ ]*\.tmp)# Describe the sinka3.sinks.k3.type = hdfsa3.sinks.k3.hdfs.path = hdfs://hadoop102: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#设置每个文件的滚动大小大概是128Ma3.sinks.k3.hdfs.rollSize = 134217700#文件的滚动与Event数量无关a3.sinks.k3.hdfs.rollCount = 0# Use a channel which buffers events in memorya3.channels.c3.type = memorya3.channels.c3.capacity = 1000a3.channels.c3.transactionCapacity = 100# Bind the source and sink to the channela3.sources.r3.channels = c3a3.sinks.k3.channel = c3
[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-dir-hdfs.conf
说明： 在使用Spooling Directory Source时
1)不要在监控目录中创建并持续修改文件 2)上传完成的文件会以.COMPLETED结尾 3)被监控文件夹每500毫秒扫描一次文件变动 3. 向upload文件夹中添加文件 在/opt/module/flume目录下创建upload文件夹[jinghang@hadoop102 flume]$ mkdir upload
向upload文件夹中添加文件
[jinghang@hadoop102 upload]$ touch jinghang.txt[jinghang@hadoop102 upload]$ touch jinghang.tmp[jinghang@hadoop102 upload]$ touch jinghang.log
-
查看HDFS上的数据
-
等待1s，再次查询upload文件夹
[jinghang@hadoop102 upload]$ ll总用量 0-rw-rw-r--. 1 jinghang jinghang 0 5月 20 22:31 jinghang.log.COMPLETED-rw-rw-r--. 1 jinghang jinghang 0 5月 20 22:31 jinghang.tmp-rw-rw-r--. 1 jinghang jinghang 0 5月 20 22:31 jinghang.txt.COMPLETED
单数据源多出口案例(选择器)
3）实现步骤： 0．准备工作 在/opt/module/flume/job目录下创建group1文件夹 [jinghang@hadoop102 job]$ cd group1/
在/opt/module/datas/目录下创建flume3文件夹
[jinghang@hadoop102 datas]$ mkdir flume3
1．创建flume-file-flume.conf
配置1个接收日志文件的source和两个channel、两个sink，分别输送给flume-flume-hdfs和flume-flume-dir。 创建配置文件并打开[jinghang@hadoop102 group1]$ touch flume-file-flume.conf[jinghang@hadoop102 group1]$ vim flume-file-flume.conf
添加如下内容
# Name the components on this agenta1.sources = r1a1.sinks = k1 k2a1.channels = c1 c2# 将数据流复制给所有channela1.sources.r1.selector.type = replicating# Describe/configure the sourcea1.sources.r1.type = execa1.sources.r1.command = tail -F /opt/module/hive/logs/hive.loga1.sources.r1.shell = /bin/bash -c# Describe the sink# sink端的avro是一个数据发送者a1.sinks.k1.type = avroa1.sinks.k1.hostname = hadoop102 a1.sinks.k1.port = 4141a1.sinks.k2.type = avroa1.sinks.k2.hostname = hadoop102a1.sinks.k2.port = 4142# Describe the channela1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100a1.channels.c2.type = memorya1.channels.c2.capacity = 1000a1.channels.c2.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1 c2a1.sinks.k1.channel = c1a1.sinks.k2.channel = c2
注：Avro是由Hadoop创始人Doug Cutting创建的一种语言无关的数据序列化和RPC框架。 注：RPC（Remote Procedure Call）—远程过程调用，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。
2．创建flume-flume-hdfs.conf 配置上级Flume输出的Source，输出是到HDFS的Sink。 创建配置文件并打开[jinghang@hadoop102 group1]$ touch flume-flume-hdfs.conf[jinghang@hadoop102 group1]$ vim flume-flume-hdfs.conf
添加如下内容
# Name the components on this agenta2.sources = r1a2.sinks = k1a2.channels = c1# Describe/configure the source# source端的avro是一个数据接收服务a2.sources.r1.type = avroa2.sources.r1.bind = hadoop102a2.sources.r1.port = 4141# Describe the sinka2.sinks.k1.type = hdfsa2.sinks.k1.hdfs.path = hdfs://hadoop102: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#设置每个文件的滚动大小大概是128Ma2.sinks.k1.hdfs.rollSize = 134217700#文件的滚动与Event数量无关a2.sinks.k1.hdfs.rollCount = 0# Describe the channela2.channels.c1.type = memorya2.channels.c1.capacity = 1000a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela2.sources.r1.channels = c1a2.sinks.k1.channel = c1
3．创建flume-flume-dir.conf
配置上级Flume输出的Source，输出是到本地目录的Sink。 创建配置文件并打开[jinghang@hadoop102 group1]$ touch flume-flume-dir.conf[jinghang@hadoop102 group1]$ vim flume-flume-dir.conf
添加如下内容
# Name the components on this agenta3.sources = r1a3.sinks = k1a3.channels = c2# Describe/configure the sourcea3.sources.r1.type = avroa3.sources.r1.bind = hadoop102a3.sources.r1.port = 4142# Describe the sinka3.sinks.k1.type = file_rolla3.sinks.k1.sink.directory = /opt/module/data/flume3# Describe the channela3.channels.c2.type = memorya3.channels.c2.capacity = 1000a3.channels.c2.transactionCapacity = 100# Bind the source and sink to the channela3.sources.r1.channels = c2a3.sinks.k1.channel = c2
提示：输出的本地目录必须是已经存在的目录，如果该目录不存在，并不会创建新的目录。
4．执行配置文件 分别开启对应配置文件：flume-flume-dir，flume-flume-hdfs，flume-file-flume。[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group1/flume-flume-dir.conf[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group1/flume-flume-hdfs.conf[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group1/flume-file-flume.conf
5．启动Hadoop和Hive
[jinghang@hadoop102 hadoop-2.7.2]$ sbin/start-dfs.sh[jinghang@hadoop103 hadoop-2.7.2]$ sbin/start-yarn.sh[jinghang@hadoop102 hive]$ bin/hivehive (default)>
6．检查HDFS上数据
7检查/opt/module/datas/flume3目录中数据
[jinghang@hadoop102 flume3]$ ll总用量 8-rw-rw-r--. 1 jinghang jinghang 5942 5月 22 00:09 1526918887550-3
单数据源多出口案例(Sink组)
[jinghang@hadoop102 job]$ cd group2/
1．创建flume-netcat-flume.conf
配置1个接收日志文件的source和1个channel、两个sink，分别输送给flume-flume-console1和flume-flume-console2。 创建配置文件并打开[jinghang@hadoop102 group2]$ touch flume-netcat-flume.conf[jinghang@hadoop102 group2]$ vim flume-netcat-flume.conf
添加如下内容
# Name the components on this agenta1.sources = r1a1.channels = c1a1.sinkgroups = g1a1.sinks = k1 k2# Describe/configure the sourcea1.sources.r1.type = netcata1.sources.r1.bind = localhosta1.sources.r1.port = 44444a1.sinkgroups.g1.processor.type = load_balancea1.sinkgroups.g1.processor.backoff = truea1.sinkgroups.g1.processor.selector = round_robina1.sinkgroups.g1.processor.selector.maxTimeOut=10000# Describe the sinka1.sinks.k1.type = avroa1.sinks.k1.hostname = hadoop102a1.sinks.k1.port = 4141a1.sinks.k2.type = avroa1.sinks.k2.hostname = hadoop102a1.sinks.k2.port = 4142# Describe the channela1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinkgroups.g1.sinks = k1 k2a1.sinks.k1.channel = c1a1.sinks.k2.channel = c1
注：Avro是由Hadoop创始人Doug Cutting创建的一种语言无关的数据序列化和RPC框架。 注：RPC（Remote Procedure Call）—远程过程调用，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。
2．创建flume-flume-console1.conf 配置上级Flume输出的Source，输出是到本地控制台。 创建配置文件并打开[jinghang@hadoop102 group2]$ touch flume-flume-console1.conf[jinghang@hadoop102 group2]$ vim flume-flume-console1.conf
添加如下内容
# Name the components on this agenta2.sources = r1a2.sinks = k1a2.channels = c1# Describe/configure the sourcea2.sources.r1.type = avroa2.sources.r1.bind = hadoop102a2.sources.r1.port = 4141# Describe the sinka2.sinks.k1.type = logger# Describe the channela2.channels.c1.type = memorya2.channels.c1.capacity = 1000a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela2.sources.r1.channels = c1a2.sinks.k1.channel = c1
3．创建flume-flume-console2.conf
配置上级Flume输出的Source，输出是到本地控制台。 创建配置文件并打开[jinghang@hadoop102 group2]$ touch flume-flume-console2.conf[jinghang@hadoop102 group2]$ vim flume-flume-console2.conf
添加如下内容
# Name the components on this agenta3.sources = r1a3.sinks = k1a3.channels = c2# Describe/configure the sourcea3.sources.r1.type = avroa3.sources.r1.bind = hadoop102a3.sources.r1.port = 4142# Describe the sinka3.sinks.k1.type = logger# Describe the channela3.channels.c2.type = memorya3.channels.c2.capacity = 1000a3.channels.c2.transactionCapacity = 100# Bind the source and sink to the channela3.sources.r1.channels = c2a3.sinks.k1.channel = c2
4．执行配置文件
分别开启对应配置文件：flume-flume-console2，flume-flume-console1，flume-netcat-flume。[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group2/flume-flume-console2.conf -Dflume.root.logger=INFO,console
[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group2/flume-flume-console1.conf -Dflume.root.logger=INFO,console
[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group2/flume-netcat-flume.conf
- 使用netcat工具向本机的44444端口发送内容 $ nc localhost 44444
- 查看Flume2及Flume3的控制台打印日志
多数据源汇总案例
3）实现步骤： 0．准备工作 分发Flume [jinghang@hadoop102 module]$ xsync flume
在hadoop102、hadoop103以及hadoop104的/opt/module/flume/job目录下创建一个group3文件夹。
[jinghang@hadoop102 job]$ mkdir group3[jinghang@hadoop103 job]$ mkdir group3[jinghang@hadoop104 job]$ mkdir group3
1．创建flume1-logger-flume.conf
配置Source用于监控hive.log文件，配置Sink输出数据到下一级Flume。 在hadoop103上创建配置文件并打开[jinghang@hadoop103 group3]$ touch flume1-logger-flume.conf[jinghang@hadoop103 group3]$ vim flume1-logger-flume.conf
添加如下内容
# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the sourcea1.sources.r1.type = execa1.sources.r1.command = tail -F /opt/module/group.loga1.sources.r1.shell = /bin/bash -c# Describe the sinka1.sinks.k1.type = avroa1.sinks.k1.hostname = hadoop104a1.sinks.k1.port = 4141# Describe the channela1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinks.k1.channel = c12．创建flume2-netcat-flume.conf配置Source监控端口44444数据流，配置Sink数据到下一级Flume：在hadoop102上创建配置文件并打开[jinghang@hadoop102 group3]$ touch flume2-netcat-flume.conf[jinghang@hadoop102 group3]$ vim flume2-netcat-flume.conf添加如下内容# Name the components on this agenta2.sources = r1a2.sinks = k1a2.channels = c1# Describe/configure the sourcea2.sources.r1.type = netcata2.sources.r1.bind = hadoop102a2.sources.r1.port = 44444# Describe the sinka2.sinks.k1.type = avroa2.sinks.k1.hostname = hadoop104a2.sinks.k1.port = 4141# Use a channel which buffers events in memorya2.channels.c1.type = memorya2.channels.c1.capacity = 1000a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela2.sources.r1.channels = c1a2.sinks.k1.channel = c1
3．创建flume3-flume-logger.conf
配置source用于接收flume1与flume2发送过来的数据流，最终合并后sink到控制台。 在hadoop104上创建配置文件并打开[jinghang@hadoop104 group3]$ touch flume3-flume-logger.conf[jinghang@hadoop104 group3]$ vim flume3-flume-logger.conf
添加如下内容
# Name the components on this agenta3.sources = r1a3.sinks = k1a3.channels = c1# Describe/configure the sourcea3.sources.r1.type = avroa3.sources.r1.bind = hadoop104a3.sources.r1.port = 4141# Describe the sink# Describe the sinka3.sinks.k1.type = logger# Describe the channela3.channels.c1.type = memorya3.channels.c1.capacity = 1000a3.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela3.sources.r1.channels = c1a3.sinks.k1.channel = c1
4．执行配置文件
分别开启对应配置文件：flume3-flume-logger.conf，flume2-netcat-flume.conf，flume1-logger-flume.conf。[jinghang@hadoop104 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group3/flume3-flume-logger.conf -Dflume.root.logger=INFO,console
[jinghang@hadoop102 flume]$
bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group3/flume2-netcat-flume.conf
[jinghang@hadoop103 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group3/flume1-logger-flume.conf
5．在hadoop103上向/opt/module目录下的group.log追加内容
[jinghang@hadoop103 module]$ echo 'hello' > group.log
6．在hadoop102上向44444端口发送数据
[jinghang@hadoop102 flume]$ telnet hadoop102 44444
3）实现步骤：
0．准备工作 分发Flume[jinghang@hadoop102 module]$ xsync flume
在hadoop102、hadoop103以及hadoop104的/opt/module/flume/job目录下创建一个group3文件夹。
[jinghang@hadoop102 job]$ mkdir group3[jinghang@hadoop103 job]$ mkdir group3[jinghang@hadoop104 job]$ mkdir group3
1．创建flume1-logger-flume.conf
配置Source用于监控hive.log文件，配置Sink输出数据到下一级Flume。 在hadoop103上创建配置文件并打开[jinghang@hadoop103 group3]$ touch flume1-logger-flume.conf[jinghang@hadoop103 group3]$ vim flume1-logger-flume.conf
添加如下内容
# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the sourcea1.sources.r1.type = execa1.sources.r1.command = tail -F /opt/module/group.loga1.sources.r1.shell = /bin/bash -c# Describe the sinka1.sinks.k1.type = avroa1.sinks.k1.hostname = hadoop104a1.sinks.k1.port = 4141# Describe the channela1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1
2．创建flume2-netcat-flume.conf
配置Source监控端口44444数据流，配置Sink数据到下一级Flume： 在hadoop102上创建配置文件并打开[jinghang@hadoop102 group3]$ touch flume2-netcat-flume.conf[jinghang@hadoop102 group3]$ vim flume2-netcat-flume.conf
添加如下内容
# Name the components on this agenta2.sources = r1a2.sinks = k1a2.channels = c1# Describe/configure the sourcea2.sources.r1.type = netcata2.sources.r1.bind = hadoop102a2.sources.r1.port = 44444# Describe the sinka2.sinks.k1.type = avroa2.sinks.k1.hostname = hadoop104a2.sinks.k1.port = 4141# Use a channel which buffers events in memorya2.channels.c1.type = memorya2.channels.c1.capacity = 1000a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela2.sources.r1.channels = c1a2.sinks.k1.channel = c13．创建flume3-flume-logger.conf配置source用于接收flume1与flume2发送过来的数据流，最终合并后sink到控制台。在hadoop104上创建配置文件并打开[jinghang@hadoop104 group3]$ touch flume3-flume-logger.conf[jinghang@hadoop104 group3]$ vim flume3-flume-logger.conf添加如下内容# Name the components on this agenta3.sources = r1a3.sinks = k1a3.channels = c1# Describe/configure the sourcea3.sources.r1.type = avroa3.sources.r1.bind = hadoop104a3.sources.r1.port = 4141# Describe the sink# Describe the sinka3.sinks.k1.type = logger# Describe the channela3.channels.c1.type = memorya3.channels.c1.capacity = 1000a3.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela3.sources.r1.channels = c1a3.sinks.k1.channel = c1
4．执行配置文件
分别开启对应配置文件：flume3-flume-logger.conf，flume2-netcat-flume.conf，flume1-logger-flume.conf。[jinghang@hadoop104 flume]$ bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group3/flume3-flume-logger.conf -Dflume.root.logger=INFO,console
[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group3/flume2-netcat-flume.conf
[jinghang@hadoop103 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group3/flume1-logger-flume.conf
5．在hadoop103上向/opt/module目录下的group.log追加内容
[jinghang@hadoop103 module]$ echo 'hello' > group.log
6．在hadoop102上向44444端口发送数据
[jinghang@hadoop102 flume]$ telnet hadoop102 44444
第4章 Flume监控之Ganglia
4.1 Ganglia的安装与部署
1) 安装httpd服务与php
[jinghang@hadoop102 flume]$ sudo yum -y install httpd php
2) 安装其他依赖
[jinghang@hadoop102 flume]$ sudo yum -y install rrdtool perl-rrdtool rrdtool-devel[jinghang@hadoop102 flume]$ sudo yum -y install apr-devel
3) 安装ganglia
[jinghang@hadoop102 flume]$ sudo rpm -Uvh http://dl.fedoraproject.org/pub/epel/6/x86_64/epel-release-6-8.noarch.rpm[jinghang@hadoop102 flume]$ sudo yum -y install ganglia-gmetad [jinghang@hadoop102 flume]$ sudo yum -y install ganglia-web[jinghang@hadoop102 flume]$ sudo yum install -y 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界面中以图表方式展现集群的运行状态下收集的多种不同指标数据。 4) 修改配置文件/etc/httpd/conf.d/ganglia.conf
[jinghang@hadoop102 flume]$ sudo vim /etc/httpd/conf.d/ganglia.conf
修改为红颜色的配置：
# Ganglia monitoring system php web frontendAlias /ganglia /usr/share/gangliaOrder deny,allow #Deny from all Allow from all # Allow from 127.0.0.1 # Allow from ::1 # Allow from .example.com
5) 修改配置文件/etc/ganglia/gmetad.conf
[jinghang@hadoop102 flume]$ sudo vim /etc/ganglia/gmetad.conf
修改为：
data_source "hadoop102" 192.168.1.102
6) 修改配置文件/etc/ganglia/gmond.conf
[jinghang@hadoop102 flume]$ sudo vim /etc/ganglia/gmond.conf
修改为：
cluster { name = "hadoop102" 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.102 port = 8649 ttl = 1}udp_recv_channel { # mcast_join = 239.2.11.71 port = 8649 bind = 192.168.1.102 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} 7) 修改配置文件/etc/selinux/config
[jinghang@hadoop102 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本次生效关闭必须重启，如果此时不想重启，可以临时生效之：
[jinghang@hadoop102 flume]$ sudo setenforce 0
5) 启动ganglia
[jinghang@hadoop102 flume]$ sudo service httpd start[jinghang@hadoop102 flume]$ sudo service gmetad start[jinghang@hadoop102 flume]$ sudo service gmond start
6) 打开网页浏览ganglia页面
尖叫提示：如果完成以上操作依然出现权限不足错误，请修改/var/lib/ganglia目录的权限： [jinghang@hadoop102 flume]$ sudo chmod -R 777 /var/lib/ganglia
4.2 操作Flume测试监控
- 修改/opt/module/flume/conf目录下的flume-env.sh配置：
JAVA_OPTS="-Dflume.monitoring.type=ganglia-Dflume.monitoring.hosts=192.168.1.102:8649-Xms100m-Xmx200m"
- 启动Flume任务
[jinghang@hadoop102 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.102:8649
- 发送数据观察ganglia监测图
[jinghang@hadoop102 flume]$ nc localhost 44444
样式如图：
图例说明： 
第5章 自定义Source
5.1 介绍
Source是负责接收数据到Flume Agent的组件。Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、tailDir、netcat、sequence generator、syslog、http、legacy。官方提供的source类型已经很多，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些source。 官方也提供了自定义source的接口： 根据官方说明自定义MySource需要继承AbstractSource类并实现Configurable和PollableSource接口。 实现相应方法： getBackOffSleepIncrement()//暂不用 getMaxBackOffSleepInterval()//暂不用 configure(Context context)//初始化context（读取配置文件内容） process()//获取数据封装成event并写入channel，这个方法将被循环调用。 使用场景：读取MySQL数据或者其他文件系统。 5.2 需求
使用flume接收数据，并给每条数据添加前缀，输出到控制台。前缀可从flume配置文件中配置。
5.2 分析 
5.3 编码 导入pom依赖 org.apache.flume flume-ng-core 1.7.0
package com.jinghang;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;public class MySource extends AbstractSource implements Configurable, PollableSource { //定义配置文件将来要读取的字段 private Long delay; private String field; //初始化配置信息 @Override public void configure(Context context) { delay = context.getLong("delay"); field = context.getString("field", "Hello!"); } @Override public Status process() throws EventDeliveryException { try { //创建事件头信息 HashMap hearderMap = new HashMap<>(); //创建事件 SimpleEvent event = new SimpleEvent(); //循环封装事件 for (int i = 0; i < 5; i++) { //给事件设置头信息 event.setHeaders(hearderMap); //给事件设置内容 event.setBody((field + i).getBytes()); //将事件写入channel getChannelProcessor().processEvent(event); Thread.sleep(delay); } } catch (Exception e) { e.printStackTrace(); return Status.BACKOFF; } return Status.READY; } @Override public long getBackOffSleepIncrement() { return 0; } @Override public long getMaxBackOffSleepInterval() { return 0; }} 5.4 测试
1）打包 将写好的代码打包，并放到flume的lib目录（/opt/module/flume）下。 2）配置文件# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the sourcea1.sources.r1.type = com.jinghang.MySourcea1.sources.r1.delay = 1000#a1.sources.r1.field = jinghang# Describe the sinka1.sinks.k1.type = logger# Use a channel which buffers events in memorya1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1
3）开启任务
[jinghang@hadoop102 flume]$ pwd/opt/module/flume[jinghang@hadoop102 flume]$ bin/flume-ng agent -c conf/ -f job/mysource.conf -n a1 -Dflume.root.logger=INFO,console
4）结果展示
第6章 自定义Sink
6.1 介绍
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的接口： 根据官方说明自定义MySink需要继承AbstractSink类并实现Configurable接口。 实现相应方法： configure(Context context)//初始化context（读取配置文件内容） process()//从Channel读取获取数据（event），这个方法将被循环调用。 使用场景：读取Channel数据写入MySQL或者其他文件系统。 6.2 需求 使用flume接收数据，并在Sink端给每条数据添加前缀和后缀，输出到控制台。前后缀可在flume任务配置文件中配置。 流程分析： 6.3 编码
package com.jinghang;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 { //创建Logger对象 private static final Logger LOG = LoggerFactory.getLogger(AbstractSink.class); private String prefix; private String suffix; @Override public Status process() throws EventDeliveryException { //声明返回值状态信息 Status status; //获取当前Sink绑定的Channel Channel ch = getChannel(); //获取事务 Transaction txn = ch.getTransaction(); //声明事件 Event event; //开启事务 txn.begin(); //读取Channel中的事件，直到读取到事件结束循环 while (true) { event = ch.take(); if (event != null) { break; } } try { //处理事件（打印） LOG.info(prefix + new String(event.getBody()) + suffix); //事务提交 txn.commit(); status = Status.READY; } catch (Exception e) { //遇到异常，事务回滚 txn.rollback(); status = Status.BACKOFF; } finally { //关闭事务 txn.close(); } return status; } @Override public void configure(Context context) { //读取配置文件内容，有默认值 prefix = context.getString("prefix", "hello:"); //读取配置文件内容，无默认值 suffix = context.getString("suffix"); }} 6.4 测试
1）打包 将写好的代码打包，并放到flume的lib目录（/opt/module/flume）下。 2）配置文件# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the sourcea1.sources.r1.type = netcata1.sources.r1.bind = localhosta1.sources.r1.port = 44444# Describe the sinka1.sinks.k1.type = com.jinghang.MySink#a1.sinks.k1.prefix = jinghang:a1.sinks.k1.suffix = :jinghang# Use a channel which buffers events in memorya1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1
3）开启任务
[jinghang@hadoop102 flume]$ pwd/opt/module/flume[jinghang@hadoop102 flume]$ bin/flume-ng agent -c conf/ -f job/mysink.conf -n a1 -Dflume.root.logger=INFO,console[jinghang@hadoop102 ~]$ nc localhost 44444helloOKjinghangOK
4）结果展示
第7章 知识扩展
7.1 常见正则表达式语法
7.2 自定义MySQLSource 7.2.1 自定义Source说明 实时监控MySQL，从MySQL中获取数据传输到HDFS或者其他存储框架，所以此时需要我们自己实现MySQLSource。 官方也提供了自定义source的接口： 官网说明： 7.2.2 自定义MySQLSource组成
图6-1 自定义MySQLSource组成
7.2.3 自定义MySQLSource步骤
根据官方说明自定义mysqlsource需要继承AbstractSource类并实现Configurable和PollableSource接口。 实现相应方法： getBackOffSleepIncrement()//暂不用 getMaxBackOffSleepInterval()//暂不用 configure(Context context)//初始化context process()//获取数据（从mysql获取数据，业务处理比较复杂，所以我们定义一个专门的类——SQLSourceHelper来处理跟mysql的交互），封装成event并写入channel，这个方法被循环调用 stop()//关闭相关的资源 PollableSource：从source中提取数据，将其发送到channel。
Configurable：实现了Configurable的任何类都含有一个context，使用context获取配置信息。 7.2.4 代码实现
1、导入pom依赖org.apache.flume flume-ng-core 1.7.0 mysql mysql-connector-java 5.1.27
2、添加配置信息
在classpath下添加jdbc.properties和log4j.properties jdbc.properties:dbDriver=com.mysql.jdbc.DriverdbUrl=jdbc:mysql://hadoop102:3306/mysqlsource?useUnicode=true&characterEncoding=utf-8dbUser=rootdbPassword=000000
log4j.properties:
#--------console-----------log4j.rootLogger=info,myconsole,myfilelog4j.appender.myconsole=org.apache.log4j.ConsoleAppenderlog4j.appender.myconsole.layout=org.apache.log4j.SimpleLayout#log4j.appender.myconsole.layout.ConversionPattern =%d [%t] %-5p [%c] - %m%n#log4j.rootLogger=error,myfilelog4j.appender.myfile=org.apache.log4j.DailyRollingFileAppenderlog4j.appender.myfile.File=/tmp/flume.loglog4j.appender.myfile.layout=org.apache.log4j.PatternLayoutlog4j.appender.myfile.layout.ConversionPattern =%d [%t] %-5p [%c] - %m%n
3、SQLSourceHelper
1) 属性说明：
2) 方法说明： 
3)代码实现： package com.jinghang.source;import org.apache.flume.Context;import org.apache.flume.conf.ConfigurationException;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.io.IOException;import java.sql.*;import java.text.ParseException;import java.util.ArrayList;import java.util.List;import java.util.Properties;public class SQLSourceHelper { private static final Logger LOG = LoggerFactory.getLogger(SQLSourceHelper.class); private int runQueryDelay, //两次查询的时间间隔 startFrom, //开始id currentIndex, //当前id recordSixe = 0, //每次查询返回结果的条数 maxRow; //每次查询的最大条数 private String table, //要操作的表 columnsToSelect, //用户传入的查询的列 customQuery, //用户传入的查询语句 query, //构建的查询语句 defaultCharsetResultSet;//编码集 //上下文，用来获取配置文件 private Context context; //为定义的变量赋值（默认值），可在flume任务的配置文件中修改 private static final int DEFAULT_QUERY_DELAY = 10000; private static final int DEFAULT_START_VALUE = 0; private static final int DEFAULT_MAX_ROWS = 2000; private static final String DEFAULT_COLUMNS_SELECT = "*"; private static final String DEFAULT_CHARSET_RESULTSET = "UTF-8"; private static Connection conn = null; private static PreparedStatement ps = null; private static String connectionURL, connectionUserName, connectionPassword; //加载静态资源 static { Properties p = new Properties(); try { p.load(SQLSourceHelper.class.getClassLoader().getResourceAsStream("jdbc.properties")); connectionURL = p.getProperty("dbUrl"); connectionUserName = p.getProperty("dbUser"); connectionPassword = p.getProperty("dbPassword"); Class.forName(p.getProperty("dbDriver")); } catch (IOException | ClassNotFoundException e) { LOG.error(e.toString()); } } //获取JDBC连接 private static Connection InitConnection(String url, String user, String pw) { try { Connection conn = DriverManager.getConnection(url, user, pw); if (conn == null) throw new SQLException(); return conn; } catch (SQLException e) { e.printStackTrace(); } return null; } //构造方法 SQLSourceHelper(Context context) throws ParseException { //初始化上下文 this.context = context; //有默认值参数：获取flume任务配置文件中的参数，读不到的采用默认值 this.columnsToSelect = context.getString("columns.to.select", DEFAULT_COLUMNS_SELECT); this.runQueryDelay = context.getInteger("run.query.delay", DEFAULT_QUERY_DELAY); this.startFrom = context.getInteger("start.from", DEFAULT_START_VALUE); this.defaultCharsetResultSet = context.getString("default.charset.resultset", DEFAULT_CHARSET_RESULTSET); //无默认值参数：获取flume任务配置文件中的参数 this.table = context.getString("table"); this.customQuery = context.getString("custom.query"); connectionURL = context.getString("connection.url"); connectionUserName = context.getString("connection.user"); connectionPassword = context.getString("connection.password"); conn = InitConnection(connectionURL, connectionUserName, connectionPassword); //校验相应的配置信息，如果没有默认值的参数也没赋值，抛出异常 checkMandatoryProperties(); //获取当前的id currentIndex = getStatusDBIndex(startFrom); //构建查询语句 query = buildQuery(); } //校验相应的配置信息（表，查询语句以及数据库连接的参数） private void checkMandatoryProperties() { if (table == null) { throw new ConfigurationException("property table not set"); } if (connectionURL == null) { throw new ConfigurationException("connection.url property not set"); } if (connectionUserName == null) { throw new ConfigurationException("connection.user property not set"); } if (connectionPassword == null) { throw new ConfigurationException("connection.password property not set"); } } //构建sql语句 private String buildQuery() { String sql = ""; //获取当前id currentIndex = getStatusDBIndex(startFrom); LOG.info(currentIndex + ""); if (customQuery == null) { sql = "SELECT " + columnsToSelect + " FROM " + table; } else { sql = customQuery; } StringBuilder execSql = new StringBuilder(sql); //以id作为offset if (!sql.contains("where")) { execSql.append(" where "); execSql.append("id").append(">").append(currentIndex); return execSql.toString(); } else { int length = execSql.toString().length(); return execSql.toString().substring(0, length - String.valueOf(currentIndex).length()) + currentIndex; } } //执行查询 List
4、MySQLSource
代码实现：package com.jinghang.source;import org.apache.flume.Context;import org.apache.flume.Event;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 org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.text.ParseException;import java.util.ArrayList;import java.util.HashMap;import java.util.List;public class SQLSource extends AbstractSource implements Configurable, PollableSource { //打印日志 private static final Logger LOG = LoggerFactory.getLogger(SQLSource.class); //定义sqlHelper private SQLSourceHelper sqlSourceHelper; @Override public long getBackOffSleepIncrement() { return 0; } @Override public long getMaxBackOffSleepInterval() { return 0; } @Override public void configure(Context context) { try { //初始化 sqlSourceHelper = new SQLSourceHelper(context); } catch (ParseException e) { e.printStackTrace(); } } @Override public Status process() throws EventDeliveryException { try { //查询数据表 List
7.2.5 测试
1、jar包准备- 将mysql驱动包放入flume的lib目录下
[jinghang@hadoop102 flume]$ cp \/opt/sorfware/mysql-libs/mysql-connector-java-5.1.27/mysql-connector-java-5.1.27-bin.jar \/opt/module/flume/lib/
- 打包项目并将jar包放入flume的lib目录下 2、配置文件准备 1）创建配置文件并打开
[jinghang@hadoop102 job]$ touch mysql.conf[jinghang@hadoop102 job]$ vim mysql.conf
2）添加如下内容
# Name the components on this agenta1.sources = r1a1.sinks = k1a1.channels = c1# Describe/configure the sourcea1.sources.r1.type = com.jinghang.source.SQLSource a1.sources.r1.connection.url = jdbc:mysql://192.168.1.102:3306/mysqlsourcea1.sources.r1.connection.user = root a1.sources.r1.connection.password = 000000 a1.sources.r1.table = student a1.sources.r1.columns.to.select = * #a1.sources.r1.incremental.column.name = id #a1.sources.r1.incremental.value = 0 a1.sources.r1.run.query.delay=5000# Describe the sinka1.sinks.k1.type = logger# Describe the channela1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channela1.sources.r1.channels = c1a1.sinks.k1.channel = c1
3、mysql表准备
1) 创建mysqlsource数据库CREATE DATABASE mysqlsource；
2) 在mysqlsource数据库下创建数据表student和元数据表flume_meta
CREATE TABLE `student` (`id` int(11) NOT NULL AUTO_INCREMENT,`name` varchar(255) NOT NULL,PRIMARY KEY (`id`));CREATE TABLE `flume_meta` (`source_tab` varchar(255) NOT NULL,`currentIndex` varchar(255) NOT NULL,PRIMARY KEY (`source_tab`));
3)向数据表中添加数据
1 zhangsan2 lisi3 wangwu4zhaoliu
4、测试并查看结果
1)任务执行[jinghang@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a1 \--conf-file job/mysql.conf -Dflume.root.logger=INFO,console
2)结果展示，如图所示：
7.3 练习 案例需求： 1）flume-1监控hive.log日志，flume-1的数据传送给flume-2，flume-2将数据追加到本地文件，同时将数据传输到flume-3。 2）flume-4监控本地另一个自己创建的文件any.txt，并将数据传送给flume-3。 3）flume-3将汇总数据写入到HDFS。 请先画出结构图，再开始编写任务脚本。 第8章 企业真实面试题（重点）
8.1 你是如何实现Flume数据传输的监控的
使用第三方框架Ganglia实时监控Flume。 8.2 Flume的Source，Sink，Channel的作用？你们Source是什么类型？ 1、作用 （1）Source组件是专门用来收集数据的，可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy （2）Channel组件对采集到的数据进行缓存，可以存放在Memory或File中。 （3）Sink组件是用于把数据发送到目的地的组件，目的地包括Hdfs、Logger、avro、thrift、ipc、file、Hbase、solr、自定义。 2、我公司采用的Source类型为： （1）监控后台日志：exec （2）监控后台产生日志的端口：netcatExec spooldir
8.3 Flume的Channel Selectors
8.4 Flume参数调优 - Source 增加Source个数（使用Tair Dir Source时可增加FileGroups个数）可以增大Source的读取数据的能力。例如：当某一个目录产生的文件过多时需要将这个文件目录拆分成多个文件目录，同时配置好多个Source 以保证Source有足够的能力获取到新产生的数据。 batchSize参数决定Source一次批量运输到Channel的event条数，适当调大这个参数可以提高Source搬运Event到Channel时的性能。
- Channel type 选择memory时Channel的性能最好，但是如果Flume进程意外挂掉可能会丢失数据。type选择file时Channel的容错性更好，但是性能上会比memory channel差。 使用file Channel时dataDirs配置多个不同盘下的目录可以提高性能。 Capacity 参数决定Channel可容纳最大的event条数。transactionCapacity 参数决定每次Source往channel里面写的最大event条数和每次Sink从channel里面读的最大event条数。transactionCapacity需要大于Source和Sink的batchSize参数。
- Sink 增加Sink的个数可以增加Sink消费event的能力。Sink也不是越多越好够用就行，过多的Sink会占用系统资源，造成系统资源不必要的浪费。 batchSize参数决定Sink一次批量从Channel读取的event条数，适当调大这个参数可以提高Sink从Channel搬出event的性能。 8.5 Flume的事务机制 Flume的事务机制（类似数据库的事务机制）：Flume使用两个独立的事务分别负责从Soucrce到Channel，以及从Channel到Sink的事件传递。比如spooling directory source 为文件的每一行创建一个事件，一旦事务中所有的事件全部传递到Channel且提交成功，那么Soucrce就将该文件标记为完成。同理，事务以类似的方式处理从Channel到Sink的传递过程，如果因为某种原因使得事件无法记录，那么事务将会回滚。且所有的事件都会保持到Channel中，等待重新传递。 8.6 Flume采集数据会丢失吗? 不会，Channel存储可以存储在File中，数据传输自身有事务。
转载地址：http://ygrzi.baihongyu.com/
你可能感兴趣的文章
Flutter UI基础 - Row、Column详解
查看>>
Flutter UI基础 - 添加背景图片
查看>>
Flutter UI基础 - 布局之Row/Column/Stack
查看>>
Flutter UI基础 - 层叠布局Stack的使用
查看>>
Flutter UI基础 - webview 使用和交互
查看>>
Flutter UI基础 - 时间选择器
查看>>
Idea - 创建Java类时，自动在文件头中添加作者和创建时间
查看>>
Docker - ASP.NET Core Docker部署
查看>>
Docker - 容器运行 .Net Core
查看>>
Django - TypeError: __init__() missing 1 required positional argument: ‘on_delete‘ 的解决办法
查看>>
Go - 解决 go get 超时问题
查看>>
Go - goose 数据库迁移工具
查看>>
SQL - SQL Server 之遍历数据集合的几种方法
查看>>
SQL - SQL Server 之处理JSON数据
查看>>
SQL - SQL Server 之ETL详解
查看>>
SQL - SQL Server 之Merge函数使用详解
查看>>
SQL - SQL Server 之WHILE循环的坑
查看>>
SQL - SQL Server中如何获取当前年，月，日，时，分，秒
查看>>
SQL - SQL Server 性能优化之SQL语句总结
查看>>
Docker - docker-compose常用命令
查看>>