Flume自定义文件命名格式(RollFileSink)

Flume自带的本地文件Sink进行存储即RollingFileSink，其主要的官网参数如下

其中sink.rollInterval表示每个多久另起一个文件，比数据以每小时(sink.rollInterval=3600)写一个文件，那么flume就会自启动起开始每小时生成一个新文件，而且文件的名称是以时间戳的的方式命名的，非常不直观友好，比如在2019-12-01 00:00:00开始那第一个文件就是flume-1575129600000（flume-倒是可以通过sink.PathManage.prefix来自定义），这样的不能一样一眼看出文件哪个时间段的，比较常用的可以为flume-2019120100之类的方式，针对这个问题对RollingFileSink进行小小的升级，以实现可以自定义文件命名格式

升级后支持的特性

支持文件名称格式以yyyyMMddHHmmSS年月日时的进行配置
支持同一个rollInterval内重启多次，文件名不会冲突

新增参数

sink.rollInterval=3600   --原有参数不变，但是意义略有不同，原先是rollInterval是自启动程序起文件间隔切换时间。现在表示文件存储时间单位间隔数，如果想要每十分为一个间隔(即如flume-2019120110 flume-2019120120 flume-2019120130。。。)，则rollInterval=600，
sink.file.name.timeFormat=yyyyMMddHH  --文件的命名方式，可以精确到秒，配合rollInterval，
sink.fileMonitor=20  --监控文件名称切换的时间，(原来这个参数是rollInterval兼任这个意义)，例如每20s看下当前时间所属时间段，如果变换就需要进行创建新文件

工程准备

首先自定义一个Flume-sink工程，将原有的RollingFileSink和PathManager拷贝过来，在pom中引入下面三个工程

<dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-sdk</artifactId><version>1.7.0</version>
</dependency>
<dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.7.0</version>
</dependency>
<dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-configuration</artifactId><version>1.7.0</version>
</dependency>

代码更新如下

RollingFileSink 类，变动部分均加上了中文解释，为节省篇幅，部分较多未动的代码有所省略，已经有所注释，请注意

public class RollingFileSink extends AbstractSink implements Configurable {private static final Logger logger = LoggerFactory.getLogger(RollingFileSink.class);//........省略了类中原有变量，主要的变量定义如下private int fileMonitor;private String fileNameTimeFormat;private PathManager pathController;private long rollInterval;//新加了一个全局变量，当前文件名称private volatile String currentFile;public RollingFileSink() {pathController = new PathManager();shouldRotate = false;}public void configure(Context context) {String directory = context.getString("sink.directory");String rollInterval = context.getString("sink.rollInterval");//.......此处有省略其他参数读取/*** 新增两个参数，文件名前缀和时间格式，如果不配置时间时间则默认是时间戳的方式*/filePrefix=context.getString("sink.file.prefix","flume-");fileNameTimeFormat=context.getString("sink.file.name.timeFormat","");if (rollInterval == null) {this.rollInterval = defaultRollInterval;} else {this.rollInterval = Long.parseLong(rollInterval);}batchSize = context.getInteger("sink.batchSize", defaultBatchSize);fileMonitor=context.getInteger("sink.fileMonitor",10);this.directory = new File(directory);if (sinkCounter == null) {sinkCounter = new SinkCounter(getName());}}//启动函数public void start() {logger.info("Starting {}...", this);sinkCounter.start();super.start();pathController.setBaseDirectory(directory);pathController.setFileNameFormat(fileNameTimeFormat);pathController.setFilePrefix(filePrefix);pathController.setRollInterval((int) rollInterval/60);int fileRoll=fileMonitor;if (rollInterval > 0) {rollService = Executors.newScheduledThreadPool(1,new ThreadFactoryBuilder().setNameFormat("rollingFileSink-roller-" +Thread.currentThread().getId() + "-%d").build());//这里跟原先不同，原先这里参数直接是rollInterval，现在换成了专门的fileMonitorrollService.scheduleAtFixedRate(new Runnable() {public void run() {File file=pathController.getCurrentFile();//在这里获取当前文件名称currentFile=file.getName();shouldRotate = true;}}, fileRoll, fileRoll, TimeUnit.SECONDS);} else {logger.info("RollInterval is not valid, file rolling will not happen.");}logger.info("RollingFileSink {} started.", getName());}//数据处理函数public Sink.Status process() throws EventDeliveryException {//获取当前时间实际应该是那个最新的文件名String newestFileName=pathController.latestFile();//如果当前文件名不是当前时间段最新的文件名，则表示需要切换文件了，关闭上一个在写的文件if (shouldRotate&&(!currentFile.equals(newestFileName))) {if (outputStream != null) {logger.info("Closing file {}", pathController.getCurrentFile());try {serializer.flush();serializer.beforeClose();outputStream.close();sinkCounter.incrementConnectionClosedCount();shouldRotate = false;} catch (IOException e) {sinkCounter.incrementConnectionFailedCount();throw new EventDeliveryException("Unable to rotate file "+ pathController.getCurrentFile() + " while delivering event", e);} finally {serializer = null;outputStream = null;}pathController.rotate();}}//打开新的文件if (outputStream == null) {File currentFile = pathController.getCurrentFile();logger.info("Opening output stream for file {}", currentFile);try {outputStream = new BufferedOutputStream(new FileOutputStream(currentFile));serializer = EventSerializerFactory.getInstance(serializerType, serializerContext, outputStream);serializer.afterCreate();sinkCounter.incrementConnectionCreatedCount();} catch (IOException e) {sinkCounter.incrementConnectionFailedCount();throw new EventDeliveryException("Failed to open file "+ pathController.getCurrentFile() + " while delivering event", e);}}Channel channel = getChannel();Transaction transaction = channel.getTransaction();Event event = null;Sink.Status result = Sink.Status.READY;String content;try {/ .........此处代码跟原先一样，为节省篇幅，此处省略} catch (Exception ex) {transaction.rollback();throw new EventDeliveryException("Failed to process transaction", ex);} finally {transaction.close();}retrn result;}@Overridepublic void stop() {//...stop()函数不变，此处代码跟原先一样，为节省篇幅，此处省略}//...省略其他set方法
}

PathManager 类

public class PathManager {private static final Logger logger = LoggerFactory.getLogger(PathManager.class);private File baseDirectory;private AtomicInteger fileIndex;private File currentFile;private String filePrefix="";private String fileNameTimeFormat="";public void setRollInterval(int rollInterval) {this.rollInterval = rollInterval;}private int rollInterval;public PathManager() {fileIndex = new AtomicInteger();}/*** 自定义文件命名格式，以时间格式如yyyyMMddHH来进行文件命名，而不是时间戳的方式* 将原先方法进行修改如下* @return*/public File nextFile() {String fileName;if(!fileNameTimeFormat.equals("")){SimpleDateFormat sdf=new SimpleDateFormat(fileNameTimeFormat);String startPrefix=sdf.format(new Date());Calendar calendar=Calendar.getInstance();int startMinute=calendar.get(Calendar.MINUTE);int startTime=(startMinute/rollInterval)*rollInterval;startPrefix=startTime<10?startPrefix+"0"+startTime:startPrefix+startTime;fileName= filePrefix+startPrefix;}else{fileName=filePrefix+getSeriesTimestamp()+"-"+fileIndex.incrementAndGet();}currentFile = new File(baseDirectory, fileName);//如果已经存在文件，则将已有文件重命名，一般用在一个时间内多次重复服务的情况if(currentFile.exists()){String newName=fileName+"-"+getRandom();currentFile.renameTo(new File(baseDirectory,newName));currentFile = new File(baseDirectory, fileName);}return currentFile;}/*** 获取当前时间点是该是格式的文件名称* @return*/public String latestFile(){SimpleDateFormat sdf=new SimpleDateFormat(fileNameTimeFormat);String startPrefix=sdf.format(new Date());Calendar calendar=Calendar.getInstance();int startMinute=calendar.get(Calendar.MINUTE);int startTime=(startMinute/rollInterval)*rollInterval;//以每个小时0为起点，rollInterval一个区间，如rollInterval=15则，文件为2019120100，2019120115、2019120130、2019120145startPrefix=startTime<10?startPrefix+"0"+startTime:startPrefix+startTime;String fileName= filePrefix+startPrefix;return fileName;}public File getCurrentFile() {if (currentFile == null) {return nextFile();}return currentFile;}public void rotate() {currentFile = null;}public File getBaseDirectory() {return baseDirectory;}public void setBaseDirectory(File baseDirectory) {this.baseDirectory = baseDirectory;}private long getSeriesTimestamp() {return System.currentTimeMillis();}public void setFilePrefix(String prefix){filePrefix=prefix;}public void setFileNameFormat(String patten){fileNameTimeFormat=patten;}public AtomicInteger getFileIndex() {return fileIndex;}//获取一个随机值public String getRandom(){return  String.valueOf((int)(1+Math.random()*(1000)));}
}