Hadoop学习（15）-- Flume 入门案例

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.* 】这样的正则表达式在实际生产中是非常必要使用的话，那么就必须通过修改源码重新编译的方式来实现了。
修改 ReliableTaildirEventReader

 if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) {
 
修改为：
 if (tf == null) {<!-- -->

修改TailFile

 if (this.inode == inode && this.path.equals(path)) {
修改为：
    if (this.inode == inode) {<!-- -->

然后将修改过的代码打包为自定义source的jar，可以直接打包taildirSource组件即可，然后替换该组件的jar。

当flume监控的日志文件被移走或删除，flume仍然在监控中，并没有释放资源，当然，在一定时间后会自动释放，这个时间根据官方文档设置默认值是120000ms。

为了避免这么长时间的资源浪费，可以把这个值调小一些。但是，官方给定的默认值为什么这么大（相对于类似超时时间都是秒单位的，而这是分钟单位）？当然不能为所欲为的把这个值改小，频繁的开关文件资源造成系统资源的浪费更应该考虑。