重复文件

Flume在修改⽂件名后会重复读取⽂件问题

Flume在修改⽂件名后会重复读取⽂件问题

问题描述：

使⽤正则表⽰监控⽂件名时，当修改⽂件名称之后，会重复读取数据。

问题场景：

在⽣产环境下，使⽤log4j打印⽇志框架时，会变更打印⽇志名称，造成flume重复读取

问题重现：

1.配置信息

#Namethecomponentsonthisagent

s=r1

=k1

ls=c1

#Describe/configurethesource

=TAILDIR

oups=f1

oups.f1=/opt/module/data/flume.*

onFile=/opt/module/flume/taildir/taildir_

#Describethesink

=logger

#Uachannelwhichbufferventsinmemory

=memory

ty=1000

ctionCapacity=100

#Bindthesourceandsinktothechannel

ls=c1

l=c1

2.启动任务

bin/flume-ngagent-na1-cconf-fconf/=INFO,console

3.测试

3.1在/opt/module/data⽬录下创建flume.开头的⽂件

3.2写⼊数据

3.3修改⽂件名

echoaaa>>

echobbb>>

123

4.查看控制台，会发现控制台上当前⽂件输出了两次。

2020-06-2415:37:34,682(SinkRunner-PollingRunner-DefaultSinkProcessor)[s(:95)]Eve

nt:{headers:{}body:616161aaa}

2020-06-2415:38:06,712(SinkRunner-PollingRunner-DefaultSinkProcessor)[s(:95)]Eve

nt:{headers:{}body:626262bbb}

2020-06-2415:38:36,719(PollableSourceRunner-TaildirSource-r1)[le(ReliableT

:290)]Openingfile:/opt/module/data/123,inode:405574,pos:0

2020-06-2415:38:36,719(SinkRunner-PollingRunner-DefaultSinkProcessor)[s(:95)]Eve

nt:{headers:{}body:616161aaa}

2020-06-2415:38:36,720(SinkRunner-PollingRunner-DefaultSinkProcessor)[s(:95)]Eve

nt:{headers:{}body:626262bbb}

解决⽅案

Flume在判断⽂件是否为新⽂件的时候会记录两个值：

indode：linux⽂件的唯⼀id

file：⽂件路径

解决⽅案⼀：协商处理

不变更⽂件名

跟公司后台⼈员协商；

让他们使⽤类似logback不更名打印⽇志框架，不要使⽤log4j会更名的打印⽇志框架。

解决⽅案⼆：修改源码

下载flume的源码包，打开flume-taildir-source项⽬⽂件。

修改地点⼀：

publicbooleanupdatePos(Stringpath,longinode,longpos)throwsIOException{

//if(==inode&&(path)){

if(==inode){

tPos(pos);

updateFilePos(pos);

("Updatedposition,file:"+path+",inode:"+inode+",pos:"+pos);

returntrue;

}

returnfal;

}

将if判断条件的&&逻辑语句后半段删除。

修改地点⼆：

publicListupdateTailFiles(booleanskipToEnd)throwsIOException{

updateTime=tTimeMillis();

ListupdatedInodes=ayList();

for(TaildirMatchertaildir:taildirCache){

Mapheaders=(eGroup());

for(Filef:chingFiles()){

longinode;

try{

inode=getInode(f);

}catch(NoSuchFileExceptione){

("Filehasbeendeletedinthemeantime:"+sage());

continue;

}

TailFiletf=(inode);

//if(tf==null||!h().equals(olutePath())){

if(tf==null){

longstartPos=skipToEnd?():0;

tf=openFile(f,headers,inode,startPos);

}el{

booleanupdated=tUpdated()

if(updated){

if(()==null){

tf=openFile(f,headers,inode,());

}

if(()<()){

("Pos"+()+"islargerthanfilesize!"

+"Restartingfrompos0,file:"+h()+",inode:"+inode);

Pos(h(),inode,0);

}

}

dTail(updated);

}

(inode,tf);

(inode);

}

}

returnupdatedInodes;

}

对于新⽣成的⽂件不再添加路径是否相同条件

编译成jar包，替换掉flume/lib⽬录下的版本可能不同