1. 监控端口数据官方案例
案例需求：首先，Flume监控本机44444端口，然后通过telnet工具向本机44444端口发送消息，最后Flume将监听的数据实时显示在控制台。
1.1 实现步骤
安装telnet工具
判断44444端口是否被占用
sudo netstat -tunlp | grep 44444
创建Flume Agent配置文件flume-telnet-logger.conf
在flume目录下创建job文件夹并进入job文件夹。
mkdir job
cd job/
在job文件夹下创建Flume Agent配置文件flume-telnet-logger.conf。
touch flume-telnet-logger.conf
在flume-telnet-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.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
开启flume监听端口
参数说明：
--conf conf/：表示配置文件存储在conf/目录- –name a1 ：表示给agent起名为a1
--conf-file job/flume-telnet.conf：flume本次启动读取的配置文件是在job文件夹下的flume-telnet.conf文件。-Dflume.root.logger==INFO,console：-D表示flume运行时动态修改flume.root.logger参数属性值，并将控制台日志打印级别设置为INFO级别。日志级别包括:log、info、warn、error。
使用telnet工具向本机的44444端口发送内容
telnet localhost 44444
在Flume监听页面观察接收数据情况
2. 实时读取本地文件到HDFS案例
案例需求：实时监控Hive日志，并上传到HDFS中
2.1 实现步骤
1. 添加Hadoop相关jar包
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拷贝到/export/server/flume/apache-flume-1.9.0-bin/lib文件夹下。
2. 创建flume-file-hdfs.conf文件
创建文件
touch flume-file-hdfs.conf
注：要想读取Linux系统中的文件，就得按照Linux命令的规则执行命令。由于Hive日志在Linux系统中所以读取文件的类型选择：exec即execute执行的意思。表示执行Linux命令来读取文件。
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 /tmp/root/hive.log
a2.sources.r2.shell = /bin/bash -c
# Describe the sink
a2.sinks.k2.type = hdfs
a2.sinks.k2.hdfs.path = hdfs://master:8020/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 = 600
#设置每个文件的滚动大小
a2.sinks.k2.hdfs.rollSize = 134217700
#文件的滚动与Event数量无关
a2.sinks.k2.hdfs.rollCount = 0
#最小冗余数
a2.sinks.k2.hdfs.minBlockReplicas = 1
# Use a channel which buffers events in memory
a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000000
a2.channels.c2.transactionCapacity = 1000
# Bind the source and sink to the channel
a2.sources.r2.channels = c2
a2.sinks.k2.channel = c2
3. 执行监控配置
bin/flume-ng agent --conf conf/ --name a2 --conf-file job/flume-file-hdfs.conf
然后执行Hive的操作，然后查看HDFS
如果没有反应，则看看flume的日志的报错信息
3. 实时读取目录文件到HDFS案例
案例需求：使用Flume监听整个目录的文件
需求分析：
3.1 实现步骤
3.1.1 创建配置文件flume-dir-hdfs.conf
创建一个文件
touch flume-dir-hdfs.conf
打开文件
a3.sources = r3
a3.sinks = k3
a3.channels = c3
# Describe/configure the source
a3.sources.r3.type = spooldir
a3.sources.r3.spoolDir = /export/server/flume/job/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://master:8020/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 = 600
#设置每个文件的滚动大小大概是128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滚动与Event数量无关
a3.sinks.k3.hdfs.rollCount = 0
#最小冗余数
a3.sinks.k3.hdfs.minBlockReplicas = 1
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000000
a3.channels.c3.transactionCapacity = 1000
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3
3.1.2 启动监控文件夹命令
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-dir-hdfs.conf
说明： 在使用Spooling Directory Source时:
- 不要在监控目录中创建并持续修改文件
- 上传完成的文件会以.COMPLETED结尾
- 被监控文件夹每500毫秒扫描一次文件变动
3.1.3 向upload文件夹中添加文件
4. 实时监控目录下多个追加文件
使用的是Taildir Source
场景：Spooldir Source 适合同步文件，但是不合适对实时追加的文件进行监护并同步，而Taildir Source适合用于监听多个实时追加的文件，并且能够实现断点续传。
4.1 实现步骤
4.1.1 创建flume-taildir-hdfs.conf文件
a3.sources = r3
a3.sinks = k3
a3.channels = c3
a3.sources.r3.type = TAILDIR
a3.sources.r3.channels = c3
a3.sources.r3.positionFile = /export/server/flume/apache-flume-1.9.0-bin/taildir_position.json
a3.sources.r3.filegroups = f1 f2
a3.sources.r3.filegroups.f1 = /export/server/flume/job/files/.*file.*
a3.sources.r3.filegroups.f2 = /export/server/flume/job/files2/.*log.*
# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://master:8020/flume/upload2/%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 = 20
#设置每个文件的滚动大小大概是128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滚动与Event数量无关
a3.sinks.k3.hdfs.rollCount = 0
#最小冗余数
a3.sinks.k3.hdfs.minBlockReplicas = 1
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000000
a3.channels.c3.transactionCapacity = 1000
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3
4.1.2 执行命令
bin/flume-ng agent --conf conf/ --name a3 --conf-file ../job/flume-taildir-hdfs.conf
4.1.3 创建文件并追加值
4.1.4 问题
-
log4j的日志文件肯定是会根据规则进行滚动的：当
*.log满了就会滚动把前文件更名为*.log.1,然后重新进行*.log文件打印。这样flume就会把*.log.1文件当作新文件，又重新读取一遍，导致重复。原因：
flume会把重命名的文件重新当作新文件读取是因为正则表达式的原因，因为重命名后的文件名仍然符合正则表达式。所以第一，重命名后的文件仍然会被flume监控；第二，flume是根据文件inode&&文件绝对路径 、文件是否为null&&文件绝对路径，这样的条件来判断是否是同一个文件这个可以看源码：下载源码，放到maven项目（注意路径名称对应），找到taildirsource的包。
确实是有重复，然后看源码：flume-taildir-source工程
ReliableTaildirEventReader类的updateTailFiles方法public List<Long> updateTailFiles(boolean skipToEnd) throws IOException { updateTime = System.currentTimeMillis(); List<Long> updatedInodes = Lists.newArrayList(); for (TaildirMatcher taildir : taildirCache) { Map<String, String> headers = headerTable.row(taildir.getFileGroup()); for (File f : taildir.getMatchingFiles()) { long inode = getInode(f); TailFile tf = tailFiles.get(inode); if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) { long startPos = skipToEnd ? f.length() : 0; tf = openFile(f, headers, inode, startPos); } else { boolean updated = tf.getLastUpdated() < f.lastModified(); if (updated) { if (tf.getRaf() == null) { tf = openFile(f, headers, inode, tf.getPos()); } if (f.length() < tf.getPos()) { logger.info("Pos " + tf.getPos() + " is larger than file size! " + "Restarting from pos 0, file: " + tf.getPath() + ", inode: " + inode); tf.updatePos(tf.getPath(), inode, 0); } } tf.setNeedTail(updated); } tailFiles.put(inode, tf); updatedInodes.add(inode); } } return updatedInodes; }重点：
for (File f : taildir.getMatchingFiles()) { long inode = getInode(f); TailFile tf = tailFiles.get(inode); if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) { long startPos = skipToEnd ? f.length() : 0; tf = openFile(f, headers, inode, startPos); }TailFile 类的 updatePos 方法：
public boolean updatePos(String path, long inode, long pos) throws IOException { <strong>if (this.inode == inode && this.path.equals(path)) {<!-- --></strong> setPos(pos); updateFilePos(pos); logger.info("Updated position, file: " + path + ", inode: " + inode + ", pos: " + pos); return true; } return false; }解决方式1：
直接修改文件监控规则，比如就如正则表达式：【.*.log.* 】这样的正则表达式当然文件由.ac.log重命名为.ac.log.1会带来重复读取的问题。而正则表达式：【.*.log】当文件由.ac.log重命名为.ac.log.1就不会被flume监控，就不会有重复读取的问题。
解决方式2：
如果类似【.*.log.* 】这样的正则表达式在实际生产中是非常必要使用的话，那么就必须通过修改源码重新编译的方式来实现了。
修改ReliableTaildirEventReaderif (tf == null || !tf.getPath().equals(f.getAbsolutePath())) { 修改为： if (tf == null) {<!-- -->修改
TailFileif (this.inode == inode && this.path.equals(path)) { 修改为： if (this.inode == inode) {<!-- -->然后将修改过的代码打包为自定义source的jar，可以直接打包taildirSource组件即可，然后替换该组件的jar。
-
当flume监控的日志文件被移走或删除，flume仍然在监控中，并没有释放资源，当然，在一定时间后会自动释放，这个时间根据官方文档设置默认值是120000ms。
为了避免这么长时间的资源浪费，可以把这个值调小一些。但是，官方给定的默认值为什么这么大（相对于类似超时时间都是秒单位的，而这是分钟单位）？当然不能为所欲为的把这个值改小，频繁的开关文件资源造成系统资源的浪费更应该考虑。
