【Flume】Flume实践之采集文件内容上传至HDFS

文章目录

• 1. 需求
• 2. 配置
• • 2.1 Source
  • 2.2 Channel
  • 2.3 Sink
  • 2.4 完整的配置代码
  • 3. 实践

    1. 需求

           使用Flume从文件夹中采集数据并上传到HDFS中。要完成这个任务就需要使用在采集数据时使用Spooling Directory Source组件；传输数据时为了保证数据没有丢失风险，使用File Channel组件；输出数据时使用HDFS Sink。

    2. 配置

           Flume各个组件的参数很多，因此通常复制官网的各组件样例程序并参照参数表进行修改。

    2.1 Source

           Source组件使用Spooling Directory Source，本次实践需要指定两个参数，分别是：

    • type参数指定组件类型；
    • spoolDir参数指定要采集数据的源文件夹。

             根据需求编写的Source配置如下：

      a1.sources.r1.type = spooldir
      a1.sources.r1.spoolDir = /data/log/studentDir
      

      2.2 Channel

             Channel组件使用File Channel，本次实践需要指定三个参数，分别是：

      • type参数指定组件类型；
      • checkpointDir参数指定存放检查点文件的目录；
      • dataDirs 表示存放数据的目录。

               根据需求编写的Channel配置如下：

        a1.channels.c1.type = file
        a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint
        a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data
        

        2.3 Sink

               Sink组件使用HDFS Sink，本次实践需要指定八个参数，分别是：

        • type参数指定组件类型；
        • path参数指定最终数据要上传到的HDFS目录地址；
        • filePrefix参数指定存放数据的文件前缀；
        • fileType参数指定存放文件的类型，共有三种：SequenceFile、DataStream和CompressedStream，其中SequenceFile在Hadoop的MapReduce任务解决小文件问题时使用过；
        • writeFormat参数指定写文件的格式，有两种：Text和Writable（默认），如果后期想使用Hive或者Impala操作这份数据的话，必须在生成数据之前设置为Text，Text表示是普通文本数据；
        • rollInterval参数指定Sink间隔多长将临时文件滚动成最终目标文件，以秒为单位，默认值为30s。如果设置成0，则表示不根据时间来滚动文件（滚动roll指的是，Sink将临时文件重命名成最终目标文件，并新打开一个临时文件来写入数据）；
        • rollSize参数指定当临时文件达到多大时(单位:bytes)，滚动成目标文件，默认值为1024。如果设置成0，则表示不根据临时文件大小来滚动文件；
        • rollCount参数指定当events数据达到该数量时候,将临时文件滚动成目标文件，默认值为10。如果设置成0,则表示不根据events数据来滚动文件。

                 根据需求编写的Sink配置如下：

          a1.sinks.k1.type = hdfs
          a1.sinks.k1.hdfs.path = hdfs://bigData01:9000/flume/studentDir
          a1.sinks.k1.hdfs.filePrefix = stu-
          a1.sinks.k1.hdfs.fileType = DataStream
          a1.sinks.k1.hdfs.writeFormat = Text
          a1.sinks.k1.hdfs.rollInterval = 3600
          a1.sinks.k1.hdfs.rollSize = 134217728
          a1.sinks.k1.hdfs.rollCount = 0
          

          2.4 完整的配置代码

                 将上述三个组件的配置完成后，再将其连接起来（设置Source后接的Channel和Sink前接的Channel），完整代码如下：

          # Name the components on this agent
          a1.sources = r1
          a1.sinks = k1
          a1.channels = c1
          # Describe/configure the source
          a1.sources.r1.type = spooldir
          a1.sources.r1.spoolDir = /data/log/studentDir
          # Describe the sink
          a1.channels.c1.type = file
          a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint
          a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data
          # Use a channel
          a1.sinks.k1.type = hdfs
          a1.sinks.k1.hdfs.path = hdfs://bigData01:9000/flume/studentDir
          a1.sinks.k1.hdfs.filePrefix = stu-
          a1.sinks.k1.hdfs.fileType = DataStream
          a1.sinks.k1.hdfs.writeFormat = Text
          a1.sinks.k1.hdfs.rollInterval = 3600
          a1.sinks.k1.hdfs.rollSize = 134217728
          a1.sinks.k1.hdfs.rollCount = 0
          # Bind the source and sink to the channel
          a1.sources.r1.channels = c1
          a1.sinks.k1.channel = c1
          

          3. 实践

                 在运行Flume之前应该先检查建立采集数据的文件夹和文件，且系统此时直接启动会报错提示找不到SequenceFile，虽然我们已经把fileType改为了DataStream，但是Flume默认还是会加载这个类。这个问题的解决思路有两种：

          • 第一种解决思路：把SequenceFile相关的jar包都拷贝到Flume的lib目录下。但后续还会报错缺少HDFS相关的jar包，再拷贝对应的jar包即可。
          • 第二种解决思路：把这个节点设置为Hadoop集群的一个客户端节点，即在这个节点上配置Hadoop的框架。

                   解决了这个问题后，在a1.sinks.k1.hdfs.path路径的机器上开启Hadoop服务框架，即可启动Flume服务采集信息并上传，运行过程如下：

            • 运行Flume

              bin/flume-ng agent --name a1 --conf conf --conf-file conf/file2hdfs.conf -Dflume.root.logger=INFO,console
              

              • 检查HDFS的上传结果

                可以看到文件后有一个.tmp的后缀，是因为当前该文件并未达到配置中设置的滚动的要求，后续的数据依旧要写入到该文件。若要去掉这个后缀那根据配置需要等待1h或者等文件达到128M，除去配置的条件外，我们也可以强制关闭这个Agent来达到目的。但需要注意一点，当我们再重启Agent后，并不会给这个文件再给加上.tmp后缀，每次停止之前都会把所有的文件解除占用状态，下次启动的时候如果有新数据，则会产生新的文件，这其实仅仅模拟了自动切文件的效果。

                [root@bigData01 soft]# hdfs dfs -ls /flume/studentDir
                Found 1 items
                -rw-r--r--   1 root supergroup         65 2023-02-03 06:23 /flume/studentDir/stu-.1675376620835.tmp
                [root@bigData01 soft]# hdfs dfs -cat /flume/studentDir/*
                jack    18      male
                jessic  20      female
                tom     17      male
                

                • 检查数据源文件

                  为了防止重复读取同一个文件，Flume会在读取过的文件加上COMPLETED后缀，再不关闭Flume的前提下再加入一个class2.dat文件，会发现该文件立即读取并写入HDFS文件中。

                  [root@bigData02 studentDir]# ll
                  total 4
                  -rw-r--r--. 1 root root 65 Feb  3 05:31 class1.dat.COMPLETED
                  [root@bigData02 studentDir]# vi class2.dat
                  [root@bigData02 studentDir]# ll
                  total 8
                  -rw-r--r--. 1 root root 65 Feb  3 05:31 class1.dat.COMPLETED
                  -rw-r--r--. 1 root root 15 Feb  3 06:26 class2.dat.COMPLETED
                  -------------------------------------------------------------------
                  [root@bigData01 soft]# hdfs dfs -cat /flume/studentDir/*
                  jack    18      male
                  jessic  20      female
                  tom     17      male
                  Chaoql	23	male