Flume数据传输事务分析[转]

本文基于ThriftSource,MemoryChannel,HdfsSink三个组件,对Flume数据传输的事务进行分析,如果使用的是其他组件,Flume事务具体的处理方式将会不同。一般情况下,用MemoryChannel就好了,我们公司用的就是这个,FileChannel速度慢,虽然提供日志级别的数据恢复,但是一般情况下,不断电MemoryChannel是不会丢数据的。

Flume提供事物操作,保证用户的数据的可靠性,主要体现在:

  • 数据在传输到下个节点时(通常是批量数据),如果接收节点出现异常,比如网络异常,则回滚这一批数据。因此有可能导致数据重发
  • 同个节点内,Source写入数据到Channel,数据在一个批次内的数据出现异常,则不写入到Channel。已接收到的部分数据直接抛弃,靠上一个节点重发数据。

编程模型

Flume在对Channel进行Put和Take操作的时候,必须要用事物包住,比如:

Channel ch = new MemoryChannel();
Transaction txn = ch.getTransaction();
//事物开始
txn.begin();
try {Event eventToStage = EventBuilder.withBody("Hello Flume!",Charset.forName("UTF-8")); //往临时缓冲区Put数据 ch.put(eventToStage); //或者ch.take() //将这些数据提交到channel中 txn.commit(); } catch (Throwable t) { txn.rollback(); if (t instanceof Error) { throw (Error)t; } } finally { txn.close(); } 

Put事务流程

Put事务可以分为以下阶段:

  • doPut:将批数据先写入临时缓冲区putList
  • doCommit:检查channel内存队列是否足够合并。
  • doRollback:channel内存队列空间不足,抛弃数据

我们从Source数据接收到写入Channel这个过程对Put事物进行分析。


ThriftSource会spawn多个Worker线程(ThriftSourceHandler)去处理数据,Worker处理数据的接口,我们只看batch批量处理这个接口:

    @Overridepublic Status appendBatch(List<ThriftFlumeEvent> events) throws TException {List<Event> flumeEvents = Lists.newArrayList();for(ThriftFlumeEvent event : events) {flumeEvents.add(EventBuilder.withBody(event.getBody(), event.getHeaders())); } //ChannelProcessor,在Source初始化的时候传进来.将数据写入对应的Channel getChannelProcessor().processEventBatch(flumeEvents); ... return Status.OK; } 

事务逻辑都在processEventBatch这个方法里:

public void processEventBatch(List<Event> events) {...//预处理每行数据,有人用来做ETL嘛events = interceptorChain.intercept(events);...//分类数据,划分不同的channel集合对应的数据 // Process required channels Transaction tx = reqChannel.getTransaction(); ... //事务开始,tx即MemoryTransaction类实例 tx.begin(); List<Event> batch = reqChannelQueue.get(reqChannel); for (Event event : batch) { // 这个put操作实际调用的是transaction.doPut reqChannel.put(event); } //提交,将数据写入Channel的队列中 tx.commit(); } catch (Throwable t) { //回滚 tx.rollback(); ... } } ... } 

每个Worker线程都拥有一个Transaction实例,保存在Channel(BasicChannelSemantics)里的ThreadLocal变量currentTransaction.

那么,事务到底做了什么?

实际上,Transaction实例包含两个双向阻塞队列LinkedBlockingDeque(感觉没必要用双向队列,每个线程写自己的putList,又不是多个线程?),分别为:

  • putList
  • takeList

对于Put事物操作,当然是只用到putList了。putList就是一个临时的缓冲区,数据会先put到putList,最后由commit方法会检查channel是否有足够的缓冲区,有则合并到channel的队列。

channel.put -> transaction.doPut:

    protected void doPut(Event event) throws InterruptedException {//计算数据字节大小 int eventByteSize = (int)Math.ceil(estimateEventSize(event)/byteCapacitySlotSize); //写入临时缓冲区putList if (!putList.offer(event)) { throw new ChannelException( "Put queue for MemoryTransaction of capacity " + putList.size() + " full, consider committing more frequently, " + "increasing capacity or increasing thread count"); } putByteCounter += eventByteSize; } 

transaction.commit:

@Overrideprotected void doCommit() throws InterruptedException { //检查channel的队列剩余大小是否足够 ... int puts = putList.size(); ... synchronized(queueLock) { if(puts > 0 ) { while(!putList.isEmpty()) { //写入到channel的队列 if(!queue.offer(putList.removeFirst())) { throw new RuntimeException("Queue add failed, this shouldn't be able to happen"); } } } //清除临时队列 putList.clear(); ... } ... } 

如果在事务期间出现异常,比如channel剩余空间不足,则rollback:

@Overrideprotected void doRollback() {...//抛弃数据,没合并到channel的内存队列 putList.clear(); ... } 

Take事务

Take事务分为以下阶段:

  • doTake:先将数据取到临时缓冲区takeList
  • 将数据发送到下一个节点
  • doCommit:如果数据全部发送成功,则清除临时缓冲区takeList
  • doRollback:数据发送过程中如果出现异常,rollback将临时缓冲区takeList中的数据归还给channel内存队列。


Sink其实是由SinkRunner线程调用Sink.process方法来了处理数据的。我们从HdfsEventSink的process方法说起,Sink类都有个process方法,用来处理传输数据的逻辑。:

public Status process() throws EventDeliveryException {...Transaction transaction = channel.getTransaction();...//事务开始transaction.begin();...for (txnEventCount = 0; txnEventCount < batchSize; txnEventCount++) { //take数据到临时缓冲区,实际调用的是transaction.doTake Event event = channel.take(); if (event == null) { break; } ... //写数据到HDFS bucketWriter.append(event); ... // flush all pending buckets before committing the transaction for (BucketWriter bucketWriter : writers) { bucketWriter.flush(); } //commit transaction.commit(); ... } catch (IOException eIO) { transaction.rollback(); ... } finally { transaction.close(); } } 

大致流程图:

接着看看channel.take,作用是将数据放到临时缓冲区,实际调用的是transaction.doTake:

protected Event doTake() throws InterruptedException {...//从channel内存队列取数据synchronized(queueLock) {event = queue.poll();}...//将数据放到临时缓冲区 takeList.put(event); ... return event; } 

接着,HDFS写线程bucketWriter将take到的数据写到HDFS,如果批数据都写完了,则要commit了:

protected void doCommit() throws InterruptedException {...takeList.clear();...
}

很简单,其实就是清空takeList而已。如果bucketWriter在写数据到HDFS的时候出现异常,则要rollback:

protected void doRollback() {int takes = takeList.size();//检查内存队列空间大小,是否足够takeList写回去 synchronized(queueLock) { Preconditions.checkState(queue.remainingCapacity() >= takeList.size(), "Not enough space in memory channel " + "queue to rollback takes. This should never happen, please report"); while(!takeList.isEmpty()) { queue.addFirst(takeList.removeLast()); } ... } ... }

转载于:https://www.cnblogs.com/whtydn/p/4384199.html

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/259134.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

最近的一些校招试题摘录

最近又参加了一些校招&#xff0c;真是马不停蹄啊。多参加考试是好的&#xff0c;可以不断发现一些新的问题。下面摘录一些我不太会的题。 1.volatile的作用是什么&#xff1f; 答案&#xff1a;volatile是类型修饰符&#xff0c;用它修饰的类型变量可能会被编译器未知的因素…

yii2中的rules验证规则

2019独角兽企业重金招聘Python工程师标准>>> Rules验证规则&#xff1a;required : 必须值验证属性||CRequiredValidator 的别名, 确保了特性不为空.[[字段名],required,requiredValue>必填值,message>提示信息];email : 邮箱验证||CEmailValidator 的别名,确…

weblogic数据源配置的问题,weblogic密码破解

weblogic 报错 please increase XXX,得知是连接池出了问题&#xff0c;查看weblogic配置&#xff0c;发现没有设置超时 查看oracle 当前session&#xff0c;可以看到连接的机器&#xff0c;用户&#xff0c;当前执行的sqlid select * from v$session; select v$sql where sql_i…

自己写的简易多任务系统---基于pic18fxxx

这个工程只是实现了最简单的OS任务调度&#xff0c;对于理解任务调度有点帮助。其实就是从UC/OS-II里面摘出来的&#xff0c;没有原来的那么复杂&#xff0c;很精简&#xff0c;但道理上是一样的。工程中的CPU.C文件时直接拿Nathan Brown写好的&#xff0c;因为关于PIC任务切换…

python语言整数类型-Python 的内置数值类型

Python 是一种敏捷的、动态类型化的、极富表现力的开源编程语言&#xff0c;可以被自由地安装到多种平台上。Python 代码是被解释的。如果您对编辑、构建和执行循环较为熟悉&#xff0c;则 Python 代码对您来说更简单。但是&#xff0c;请不要搞错&#xff1a;Python 器可以是简…

滴滴出行2016校招编程题

1. 给定一个m*n的数组&#xff08;m,n>2,数组值>0&#xff09;&#xff0c;要求选出和最大的子2*2数组。例如&#xff1a; 1 2 3 4 5 6 7 8 9 显然和最大的2*2子数组是5 6;8 9.下面完成这个功能。 Input: &#xff08;m*n的数组&#xff09; 1 2 3 ; 4 5 6 ; 7 8 9 …

每天一个linux命令(22):find 命令的参数详解

find一些常用参数的一些常用实例和一些具体用法和注意事项。 1&#xff0e;使用name选项&#xff1a; 文件名选项是find命令最常用的选项&#xff0c;要么单独使用该选项&#xff0c;要么和其他选项一起使用。 可以使用某种文件名模式来匹配文件&#xff0c;记住要用引号将文件…

(WPF) DataGrid之绑定

通过ObservableCollection 绑定到 DataGrid. 1. 前台Xaml. <DataGrid x:Name"dgMeasurements"HorizontalAlignment"Left"Margin"10,69,0,10"ItemsSource"{Binding}"AutoGenerateColumns"False"Width"370">…

程序=数据结构+算法

这句名言&#xff0c;我现在品来很有感觉&#xff0c;看看uc/os-II里面那些就绪表、查找最高优先级任务等等&#xff0c;算法设计的非常巧妙&#xff0c;整个OS都是围绕着OS_TCB来运转的&#xff0c;任务需要通信&#xff0c;那就在建立个OS_EVENT&#xff0c;通过.*OSTCBEvent…

去哪笔试两题

1&#xff0c;a是一个有序数组&#xff0c;但经过向右移动数位&#xff0c;现在预在a中查找元素key的位置&#xff0c;如不存在&#xff0c;返回0。例如a[5,6.7.8,1,2,3,4]. 实现&#xff1a; 1 #quna12 def findPos(a,key):3 mina[0];4 for i in range(len(a)):5 …

MySQL5.6主从复制搭建基于日志(binlog)

什么是MySQL主从复制 简单来说&#xff0c;就是保证主SQL&#xff08;Master&#xff09;和从SQL&#xff08;Slave&#xff09;的数据是一致性的&#xff0c;向Master插入数据后&#xff0c;Slave会自动从Master把修改的数据同步过来&#xff08;有一定的延迟&#xff09;&…

opengl 如何加阴影_动漫嘴唇厚涂如何绘制?厚涂嘴唇正确画法

动漫嘴唇厚涂如何绘制&#xff1f;厚涂嘴唇正确画法&#xff01;嘴巴怎么画&#xff1f;画嘴巴真的很考验一个画师功力&#xff0c;好看的嘴巴生动而丰满&#xff0c;可以给整幅画作添上亮点&#xff0c;而画的不好的嘴巴呢&#xff0c;就容易把画面整体的风格打破。那么零基础…

位运算

我们复习一下位运算&#xff0c;这里介绍一下(& ,|, ^)的用途。 按位与 ------------& 规则&#xff1a; 0&00 0&10 1&0 0 1&11 &#xff08; 两位为1&#xff0c;才是1&#xff09;作用&#xff1a; 清零与保位。通常用来将特定的位清零&…

详解JMeter函数和变量

详解JMeter函数和变量&#xff08;1&#xff09; JMeter函数可以被认为是某种特殊的变量&#xff0c;它们可以被采样器或者其他测试元件所引用。函数调用的语法如下&#xff1a; ${__functionName(var1,var2,var3)} 其中&#xff0c;__functionName匹配被调用的函数名称。用圆括…

信号反射

突然想起来前几天调试CAN通讯的时候出现的BUG&#xff0c;那就是传说中的“信号反射”&#xff0c;也有称“振铃”的。错误刚出现的时候没有意识过来&#xff0c;还说怎么出现重复出现这么多条消息呢&#xff1f;光在书本上看到过这个概念&#xff0c;没有“实物”与之对应起来…

hdu 5199 map或二分或哈希

题目描述&#xff1a;给出n棵树的高度&#xff0c;每棵树上都站着一只鸟&#xff0c;枪手Jack站在最左边那棵树的左边对鸟进行射击&#xff0c;当Jack在高度为H的地方向右发射一颗子弹的时候&#xff0c;高度为H的树上的鸟儿就会掉落&#xff08;注&#xff1a;其他树上的鸟儿不…

数字电路实验怎么接线视频讲解_家庭影院中音箱、功放、投影机、4K播放机不知道怎么连接?手把手教你...

家庭影院中音箱、功放、投影机、4K播放机不知道怎么连接&#xff1f;手把手教你有不少用户收到从家庭影院器材之后&#xff0c;表示完全不会连接。翻看说明书也觉得头大&#xff0c;知识太多&#xff0c;然而却很难找到要点。今天主要跟大家讲讲如何连接音箱、功放、投影机和影…

.NET开发过程中的全文索引使用技巧之Solr

前言&#xff1a;相信许多人都听说过.net开发过程中基于Lucene.net实现的全文索引&#xff0c;而Solr是一个高性能&#xff0c;基于Lucene的全文搜索服务器。同时对其进行了扩展&#xff0c;提供了比Lucene更为丰富的查询语言&#xff0c;同时实现了可配置、可扩展并对查询性能…

关于字符的读入与输出

在笔试中&#xff0c;经常见到字符的读入与输出的题目。逆序打印输入时最常见、最基本的考题&#xff0c;复杂点的就是统计单词、逆序打印单词之类的。难点是如何判断输入的结束&#xff0c;如果用getchar函数&#xff0c;其输入结束符为EOF&#xff08;其打印值为-1&#xff0…

修正discuz发帖首次换行无效的问题

找遍了百度和google都没有解决方案&#xff0c;连discuz官方都没有出来解决&#xff0c;至今其官网仍有这个问题。 那就自己动手解决吧&#xff0c;顺手打个补丁。虽然走了小路&#xff0c;但是能解决问题。 解决方案&#xff1a;修改static/js/bbcode.js 找到 html2bbcode()方…