使用Flume监听整个目录的实时追加文件，并上传至HDFS

左手的ㄟ右手 2022-10-16 10:28 222阅读 0赞

**案例需求**：使用Flume监听整个目录的实时追加文件，并上传至HDFS  
**需求分析**:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
**实现步骤：**  
（1）创建配置文件flume-taildir-hdfs.conf  
创建一个文件

vim flume-taildir-hdfs.conf

添加如下内容

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = TAILDIR
    a1.sources.r1.positionFile = /export/servers/flume/tail_dir.json
    a1.sources.r1.filegroups = f1 f2
    a1.sources.r1.filegroups.f1 = /export/servers/flume/files/.*file.*
    a1.sources.r1.filegroups.f2 = /export/servers/flume/files/.*log.*
    
    # Describe the sink
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://hadoop102:8020/flume/upload2/%Y%m%d/%H
    #上传文件的前缀
    a1.sinks.k1.hdfs.filePrefix = upload-
    #是否按照时间滚动文件夹
    a1.sinks.k1.hdfs.round = true
    #多少时间单位创建一个新的文件夹
    a1.sinks.k1.hdfs.roundValue = 1
    #重新定义时间单位
    a1.sinks.k1.hdfs.roundUnit = hour
    #是否使用本地时间戳
    a1.sinks.k1.hdfs.useLocalTimeStamp = true
    #积攒多少个Event才flush到HDFS一次
    a1.sinks.k1.hdfs.batchSize = 100
    #设置文件类型，可支持压缩
    a1.sinks.k1.hdfs.fileType = DataStream
    #多久生成一个新的文件
    a1.sinks.k1.hdfs.rollInterval = 60
    #设置每个文件的滚动大小大概是128M
    a1.sinks.k1.hdfs.rollSize = 134217700
    #文件的滚动与Event数量无关
    a1.sinks.k1.hdfs.rollCount = 0
    
    # 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

（2）启动监控文件夹命令

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

（3）向files文件夹中追加内容  
在/export/servers/flume目录下创建files文件夹

mkdir files

向files文件夹中添加文件

echo hello >> file1.txt
    echo aa >> file2.txt

（4）查看HDFS上的数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]

Taildir说明：  
Taildir Source维护了一个json格式的position File，其会定期的往position File中更新每个文件读取到的最新的位置，因此能够实现断点续传。Position File的格式如下：

{ "inode":2496272,"pos":12,"file":"/export/servers/flume/files/file1.txt"}
    { "inode":2496275,"pos":12,"file":"/export/servers/flume/files/file2.txt"}

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: /images/20221014/fa2ee700f2f2473f95dee3e0e439ace9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]: /images/20221014/b8626a4bf31d4dcf9dc3ec1a0541962b.png