- 浏览: 124371 次
- 性别:
- 来自: 深圳
文章分类
最新评论
SpoolDirectorySource使用及源码分析
Spooling Directory Source简介
Spooling Directory Source可以获取硬盘上“spooling”目录的数据,这个Source将监视指定目录是否有新文件,如果有新文件的话,就解析这个新文件。事件的解析逻辑是可插拔的。在文件的内容所有的都读取到Channel之后,Spooling Directory Source会重名或者是删除该文件以表示文件已经读取完成。不像Exec Source,这个Source是可靠的,且不会丢失数据。即使Flume重启或者被Kill。但是需要注意如下两点:
1,如果文件在放入spooling目录之后还在写,那么Flume会打印错误日志,并且停止处理该文件。
2,如果文件之后重复使用,Flume将打印错误日志,并且停止处理。
为了避免以上问题,我们可以使用唯一的标识符来命令文件,例如:时间戳。
尽管这个Source是可靠的,但是如果下游发生故障,也会导致Event重复,这种情况就需要通过Flume的其他组件提供保障了。
channels | – |
|
type | – |
组件名:spooldir .
|
spoolDir | – | 读取文件的目录。 |
fileSuffix | .COMPLETED | Spooling读取过的文件,添加的后缀。 |
deletePolicy | never |
完成后的文件是否删除。never:不删除 或immediate:立即删除
|
fileHeader | false | 是不把路径加入到Heander |
fileHeaderKey | file | 路径加入到Header的Key是什么 |
basenameHeader | false | 是不把文件名加入到Heander |
basenameHeaderKey | basename | 文件名加入到Header的Key是什么 |
ignorePattern | ^$ | 采用正则表达是去过滤一些文件。只有符合正则表达式的文件才会被使用。 |
trackerDir | .flumespool | 被处理文件的元数据的存储目录,如果不是绝对路径,就被会解析到spoolDir目录下。 |
consumeOrder | oldest |
消费spooling目录文件的规则,分别有:oldest,youngest和random。在oldest 和 youngest的情况下, 通过文件的最后修改时间来比较文件。如果最后修改时间相同,就根据字典的序列从小开始。在随机的情况 下,就随意读取文件。如果文件列表很长,采用oldest/youngest可能会很慢,因为用oldest/youngest要 扫描文件。但是如果采用random的话,就可能造成新的文件消耗的很快,老的文件一直都没有被消费。 |
maxBackoff | 4000 | 如果Channel已经满了,那么该Source连续尝试写入该Channel的最长时间(单位:毫秒)。 |
batchSize | 100 | 批量传输到Channel的粒度。 |
inputCharset | UTF-8 | 字符集 |
decodeErrorPolicy |
FAIL
|
在文件中有不可解析的字符时的解析策略。FAIL : 抛出一个异常,并且不能解析该文件。REPLACE :
取代不可解析的字符,通常用Unicode U+FFFD. IGNORE : 丢弃不可能解析字符序列。
|
deserializer |
LINE
|
自定序列化的方式,自定的话,必须实现EventDeserializer.Builder .
|
deserializer.* |
|
|
bufferMaxLines | – | 已废弃。 |
bufferMaxLineLength | 5000 | (不推荐使用) 一行中最大的长度,可以使用deserializer.maxLineLength代替。 |
selector.type | replicating | replicating(复制) 或 multiplexing(复用) |
selector.* |
|
取决于selector.type的值 |
interceptors | – | 空格分割的interceptor列表。 |
interceptors.* |
|
|
SpoolDirectorySource示例
读取文件写入到file_roll中
a1.sources = source1 a1.sinks = sink1 a1.channels = channel1 #resources a1.sources.source1.type = spooldir a1.sources.source1.channels = channel1 a1.sources.source1.spoolDir = E:\\home\\spooling a1.sources.source1.fileHeader = true a1.sources.source1.fileHeaderKey = fishfile a1.sources.source1.basenameHeader = true a1.sources.source1.basenameHeaderKey = fishbasename a1.sinks.sink1.type = file_roll a1.sinks.sink1.sink.directory = E:\\home\\file_roll a1.sinks.sink1.sink.rollInterval = 300 a1.sinks.sink1.sink.serializer = TEXT a1.sinks.sink1.sink.batchSize = 100 a1.channels.channel1.type = memory a1.channels.channel1.capacity = 1000 a1.channels.channel1.transactionCapacity = 100 a1.sources.source1.channels = channel1 a1.sinks.sink1.channel = channel1
SpoolDirectorySource源码分析
一,调用configure(Context context)方法初始化:
@Override public synchronized void configure(Context context) { //spool目录 spoolDirectory = context.getString(SPOOL_DIRECTORY); Preconditions.checkState(spoolDirectory != null, "Configuration must specify a spooling directory"); //完成后的文件后缀 completedSuffix = context.getString(SPOOLED_FILE_SUFFIX, DEFAULT_SPOOLED_FILE_SUFFIX); //删除策略,never:不删除 或 immediate:立即删除 deletePolicy = context.getString(DELETE_POLICY, DEFAULT_DELETE_POLICY); //以下四个参数是是否在header中加入文件名和文件路径。 fileHeader = context.getBoolean(FILENAME_HEADER, DEFAULT_FILE_HEADER); fileHeaderKey = context.getString(FILENAME_HEADER_KEY, DEFAULT_FILENAME_HEADER_KEY); basenameHeader = context.getBoolean(BASENAME_HEADER, DEFAULT_BASENAME_HEADER); basenameHeaderKey = context.getString(BASENAME_HEADER_KEY, DEFAULT_BASENAME_HEADER_KEY); //批量处理的数量 batchSize = context.getInteger(BATCH_SIZE, DEFAULT_BATCH_SIZE); //字符集 inputCharset = context.getString(INPUT_CHARSET, DEFAULT_INPUT_CHARSET); //在文件中有不可解析的字符时的解析策略 decodeErrorPolicy = DecodeErrorPolicy.valueOf( context.getString(DECODE_ERROR_POLICY, DEFAULT_DECODE_ERROR_POLICY) .toUpperCase(Locale.ENGLISH)); //过滤文件的正则表达式 ignorePattern = context.getString(IGNORE_PAT, DEFAULT_IGNORE_PAT); //被处理文件的元数据的存储目录 trackerDirPath = context.getString(TRACKER_DIR, DEFAULT_TRACKER_DIR); //序列化 deserializerType = context.getString(DESERIALIZER, DEFAULT_DESERIALIZER); deserializerContext = new Context(context.getSubProperties(DESERIALIZER + ".")); //消费spooling目录文件的规则 consumeOrder = ConsumeOrder.valueOf(context.getString(CONSUME_ORDER, DEFAULT_CONSUME_ORDER.toString()).toUpperCase(Locale.ENGLISH)); // "Hack" to support backwards compatibility with previous generation of // spooling directory source, which did not support deserializers Integer bufferMaxLineLength = context.getInteger(BUFFER_MAX_LINE_LENGTH); if (bufferMaxLineLength != null && deserializerType != null && deserializerType.equalsIgnoreCase(DEFAULT_DESERIALIZER)) { deserializerContext.put(LineDeserializer.MAXLINE_KEY, bufferMaxLineLength.toString()); } maxBackoff = context.getInteger(MAX_BACKOFF, DEFAULT_MAX_BACKOFF); if (sourceCounter == null) { sourceCounter = new SourceCounter(getName()); } }
start方法:
@Override public synchronized void start() { logger.info("SpoolDirectorySource source starting with directory: {}", spoolDirectory); executor = Executors.newSingleThreadScheduledExecutor(); File directory = new File(spoolDirectory); //构建ReliableSpoolingFileEventReader对象 try { reader = new ReliableSpoolingFileEventReader.Builder() .spoolDirectory(directory) .completedSuffix(completedSuffix) .ignorePattern(ignorePattern) .trackerDirPath(trackerDirPath) .annotateFileName(fileHeader) .fileNameHeader(fileHeaderKey) .annotateBaseName(basenameHeader) .baseNameHeader(basenameHeaderKey) .deserializerType(deserializerType) .deserializerContext(deserializerContext) .deletePolicy(deletePolicy) .inputCharset(inputCharset) .decodeErrorPolicy(decodeErrorPolicy) .consumeOrder(consumeOrder) .build(); } catch (IOException ioe) { throw new FlumeException("Error instantiating spooling event parser", ioe); } //构建SpoolDirectoryRunnable线程。 Runnable runner = new SpoolDirectoryRunnable(reader, sourceCounter); //每隔POLL_DELAY_MS(500ms)执行以下SpoolDirectoryRunnable线程。 executor.scheduleWithFixedDelay( runner, 0, POLL_DELAY_MS, TimeUnit.MILLISECONDS); super.start(); logger.debug("SpoolDirectorySource source started"); sourceCounter.start(); }构建ReliableSpoolingFileEventReader类,构造方法功能:
1,spooling目录是否存在,是否是目录
2,通过创建零时文件测试spooling目录的权限。
3,创建trackerDir目录和.flumespool-main.meta文件
SpoolDirectoryRunnable线程,主要用于发送和读取Event:
private class SpoolDirectoryRunnable implements Runnable { private ReliableSpoolingFileEventReader reader; private SourceCounter sourceCounter; public SpoolDirectoryRunnable(ReliableSpoolingFileEventReader reader, SourceCounter sourceCounter) { this.reader = reader; this.sourceCounter = sourceCounter; } @Override public void run() { int backoffInterval = 250; try { while (!Thread.interrupted()) { //ReliableSpoolingFileEventReader读取batchSize大小的Event List<Event> events = reader.readEvents(batchSize); if (events.isEmpty()) { break; } //统计 sourceCounter.addToEventReceivedCount(events.size()); sourceCounter.incrementAppendBatchReceivedCount(); try { //将Event数组发送到Channel getChannelProcessor().processEventBatch(events); //commit会记录最后一次读取的行数,以便下次知道从哪里开始读 reader.commit(); } catch (ChannelException ex) { //ChannelProcessor批量提交Event出错,会抛出ChannelException异常,此时reader.commit是没有执行的 //所以在接下来的continue后,继续通过reader读取文件的话,还是从原来的位置读取,以保证数据不会丢失。 logger.warn("The channel is full, and cannot write data now. The " + "source will try again after " + String.valueOf(backoffInterval) + " milliseconds"); hitChannelException = true; if (backoff) { TimeUnit.MILLISECONDS.sleep(backoffInterval); backoffInterval = backoffInterval << 1; backoffInterval = backoffInterval >= maxBackoff ? maxBackoff : backoffInterval; } continue; } backoffInterval = 250; sourceCounter.addToEventAcceptedCount(events.size()); sourceCounter.incrementAppendBatchAcceptedCount(); } } catch (Throwable t) { logger.error("FATAL: " + SpoolDirectorySource.this.toString() + ": " + "Uncaught exception in SpoolDirectorySource thread. " + "Restart or reconfigure Flume to continue processing.", t); hasFatalError = true; Throwables.propagate(t); } } }ReliableSpoolingFileEventReader读取Event:
public List<Event> readEvents(int numEvents) throws IOException { //committed初始化为true if (!committed) { if (!currentFile.isPresent()) { throw new IllegalStateException("File should not roll when " + "commit is outstanding."); } logger.info("Last read was never committed - resetting mark position."); //正常情况下,会在SpoolDirectorySource类中记录读取的字节数之后,将commited设置为true //没有设置为true,可能是因为发送到Channel异常了,调用下面reset方法可以保证数据不丢失。 currentFile.get().getDeserializer().reset(); } else { // Check if new files have arrived since last call if (!currentFile.isPresent()) { //读取文件,读取文件过程中使用FileFilter过滤掉completedSuffix后缀的文件,然后根据消费文件的规则(consumeOrder)去消费文件。 currentFile = getNextFile(); } // Return empty list if no new files if (!currentFile.isPresent()) { return Collections.emptyList(); } } EventDeserializer des = currentFile.get().getDeserializer(); //根据序列化类读取Event List<Event> events = des.readEvents(numEvents); /* It's possible that the last read took us just up to a file boundary. * If so, try to roll to the next file, if there is one. * Loop until events is not empty or there is no next file in case of 0 byte files */ while (events.isEmpty()) { logger.info("Last read took us just up to a file boundary. Rolling to the next file, if there is one."); retireCurrentFile(); currentFile = getNextFile(); if (!currentFile.isPresent()) { return Collections.emptyList(); } events = currentFile.get().getDeserializer().readEvents(numEvents); } //添加<span style="font-family:Microsoft Yahei;">文件路径到Header</span> if (annotateFileName) { String filename = currentFile.get().getFile().getAbsolutePath(); for (Event event : events) { event.getHeaders().put(fileNameHeader, filename); } } //添加文件名到Header if (annotateBaseName) { String basename = currentFile.get().getFile().getName(); for (Event event : events) { event.getHeaders().put(baseNameHeader, basename); } } committed = false; lastFileRead = currentFile; return events; }消费文件的规则,以OLDEST为例:
for (File candidateFile: candidateFiles) { //已选择文件的最后修改时间,减去文件列表取的文件最后修改时间 long compare = selectedFile.lastModified() - candidateFile.lastModified(); if (compare == 0) { // ts is same pick smallest lexicographically. //时间一样就根据字典序列排序 selectedFile = smallerLexicographical(selectedFile, candidateFile); } else if (compare > 0) { // candidate is older (cand-ts < selec-ts). selectedFile = candidateFile; } }
在ReliableSpoolingFileEventReader类读取Events之后,调用commit方法提交,这里有个问题,batchSize为200,我们每次也就读取200行,那么SpoolingDirectorySource是如何标记我们读到文件的那个位置的呢?
其实SpoolingDirectorySource在commit时,会调用EventDeserializer的mark方法标记此时读取在文件的什么位置。源码如下:
public synchronized void storePosition(long position) throws IOException { metaCache.setOffset(position); writer.append(metaCache); writer.sync(); writer.flush(); }上面的方法会把文件的读取到什么位置记录到.flumespool\.flumespool-main.meta(默认情况下)文件中。
public void commit() throws IOException { if (!committed && currentFile.isPresent()) { currentFile.get().getDeserializer().mark(); //mark成功后,将committed设置为true committed = true; } }
相关推荐
杭州电子科技大学,期末考试资料,计算机专业期末考试试卷,试卷及答案,数据结构。
重庆大学期末考试试卷,重大期末考试试题,试题及答案
年会班会资料,节目策划,游戏策划,策划案,策划方案,活动方案,筹办,公司年会,开场白,主持人,策划主题,主持词,小游戏。
基于Django框架的博客系统.zip
【基于Springboot+Vue的Java毕业设计】校园服务平台项目实战(源码+录像演示+说明).rar 【项目技术】 开发语言:Java 框架:Spingboot+vue 架构:B/S 数据库:mysql 【演示视频-编号:321】 https://pan.quark.cn/s/8dea014f4d36 【实现功能】 系统可以提供信息显示和相应服务,其管理员增删改查接单员和接单员资料,审核接单员预订订单,查看订单评价和评分,通过留言功能回复用户提问。
Excel数据看板,Excel办公模板,Excel模板下载,Excel数据统计,数据展示
PTS技术内含数据集.zip
工作总结,新年计划,岗位总结,工作汇报,个人总结,述职报告,范文下载,新年总结,新建计划。
重庆大学期末考试试卷,重大期末考试试题,试题及答案
1.版本:matlab2014/2019a/2021a 2.附赠案例数据可直接运行matlab程序。 3.代码特点:参数化编程、参数可方便更改、代码编程思路清晰、注释明细。 4.适用对象:计算机,电子信息工程、数学等专业的大学生课程设计、期末大作业和毕业设计。
空间复用的MIMO系统的信号检测OSIC检测方法内含数据集.zip
该毕业设计使用了比较受前端欢迎的vue框架技术,是一个纯前端的项目,没有使用后端java程序和数据库。如果你刚学完vue相关的知识,可以通过该项目源码对vue加以巩固。现在流行无纸化办公,该项目主要解决了问卷调查相关的问题。 主要功能包括:问卷发布,数据查看。其中数据查看以图表的方式展现,便于使用者查看调查结果。
工作总结,新年计划,岗位总结,工作汇报,个人总结,述职报告,范文下载,新年总结,新建计划。
空间复用的MIMO系统的信号检测MIMO系统的QRM-MLD检测器的软判决内含数据集.zip
重庆大学期末考试试卷,重大期末考试试题,试题及答案
数字政府建设实施方案建议.rar文件是一个综合性的数字化转型解决方案精品资料,旨在为政府部门提供一个详细的蓝图和指导方针,以推进公共服务的数字化进程。这个方案强调了数字化在提高政府效率、透明度以及民众参与度方面的重要性,并提出了具体的实施步骤,包括基础设施升级、数据管理、在线服务平台建设、网络安全保障等关键要素。通过整合现代信息技术,如云计算、大数据分析、人工智能和移动互联,本方案不仅致力于优化内部工作流程,还着眼于提升公众与政府的互动体验。它涵盖了从政策制定到技术执行的各个层面,确保了数字政府建设的全面性和系统性,同时考虑到了可持续性和未来的技术演进。这份资料是任何希望实现现代化治理和服务的政府机构不可或缺的参考资源,有助于构建一个更加智能化、高效和公民友好的数字政府环境。重新回答||
一款成人用品线上零售商城JAVA源码,支持h5+安卓+ios+微信小程序, 在线销售男士、女士避孕用品、情趣内衣和玩具类等两性健康产品, 客户有自己的货源为了扩大渠道,方便客户购买,开发本款小程序。 产品分类清晰明了,首页需突出主打产品,页面展示产品清楚明了; 购买流程不要繁琐;后期方便用户查看订单情况。有促销信息及优惠券可以提高刺激消费者
工作总结,新年计划,岗位总结,工作汇报,个人总结,述职报告,范文下载,新年总结,新建计划。
1.版本:matlab2014/2019a/2021a 2.附赠案例数据可直接运行matlab程序。 3.代码特点:参数化编程、参数可方便更改、代码编程思路清晰、注释明细。 4.适用对象:计算机,电子信息工程、数学等专业的大学生课程设计、期末大作业和毕业设计。