本文基于ThriftSource,MemoryChannel,HdfsSink三個組件，對Flume數(shù)據(jù)傳輸?shù)氖聞?wù)進行分析，如果使用的是其他組件，Flume事務(wù)具體的處理方式將會不同。一般情況下，用MemoryChannel就好了，我們公司用的就是這個，FileChannel速度慢，雖然提供日志級別的數(shù)據(jù)恢復(fù)，但是一般情況下，不斷電MemoryChannel是不會丟數(shù)據(jù)的。

Flume提供事物操作，保證用戶的數(shù)據(jù)的可靠性，主要體現(xiàn)在：

• 數(shù)據(jù)在傳輸?shù)较聜€節(jié)點時(通常是批量數(shù)據(jù))，如果接收節(jié)點出現(xiàn)異常，比如網(wǎng)絡(luò)異常，則回滾這一批數(shù)據(jù)。因此有可能導(dǎo)致數(shù)據(jù)重發(fā)

• 同個節(jié)點內(nèi)，Source寫入數(shù)據(jù)到Channel,數(shù)據(jù)在一個批次內(nèi)的數(shù)據(jù)出現(xiàn)異常，則不寫入到Channel。已接收到的部分數(shù)據(jù)直接拋棄，靠上一個節(jié)點重發(fā)數(shù)據(jù)。

編程模型

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")); //往臨時緩沖區(qū)Put數(shù)據(jù) ch.put(eventToStage); //或者ch.take() //將這些數(shù)據(jù)提交到channel中 txn.commit(); } catch (Throwable t) { txn.rollback(); if (t instanceof Error) { throw (Error)t; } } finally { txn.close(); }

Put事務(wù)流程

Put事務(wù)可以分為以下階段：

• doPut:將批數(shù)據(jù)先寫入臨時緩沖區(qū)putList
• doCommit:檢查channel內(nèi)存隊列是否足夠合并。
• doRollback:channel內(nèi)存隊列空間不足，拋棄數(shù)據(jù)

我們從Source數(shù)據(jù)接收到寫入Channel這個過程對Put事物進行分析。

ThriftSource會spawn多個Worker線程(ThriftSourceHandler)去處理數(shù)據(jù)，Worker處理數(shù)據(jù)的接口，我們只看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初始化的時候傳進來.將數(shù)據(jù)寫入對應(yīng)的Channel getChannelProcessor().processEventBatch(flumeEvents); ... return Status.OK; }

事務(wù)邏輯都在processEventBatch這個方法里：

public void processEventBatch(List<Event> events) {...//預(yù)處理每行數(shù)據(jù)，有人用來做ETL嘛events = interceptorChain.intercept(events);...//分類數(shù)據(jù)，劃分不同的channel集合對應(yīng)的數(shù)據(jù) // Process required channels Transaction tx = reqChannel.getTransaction(); ... //事務(wù)開始，tx即MemoryTransaction類實例 tx.begin(); List<Event> batch = reqChannelQueue.get(reqChannel); for (Event event : batch) { // 這個put操作實際調(diào)用的是transaction.doPut reqChannel.put(event); } //提交，將數(shù)據(jù)寫入Channel的隊列中 tx.commit(); } catch (Throwable t) { //回滾 tx.rollback(); ... } } ... }

每個Worker線程都擁有一個Transaction實例，保存在Channel(BasicChannelSemantics)里的ThreadLocal變量currentTransaction.

那么，事務(wù)到底做了什么？

實際上，Transaction實例包含兩個雙向阻塞隊列LinkedBlockingDeque(感覺沒必要用雙向隊列，每個線程寫自己的putList，又不是多個線程？),分別為：

• putList
• takeList

對于Put事物操作，當(dāng)然是只用到putList了。putList就是一個臨時的緩沖區(qū)，數(shù)據(jù)會先put到putList,最后由commit方法會檢查channel是否有足夠的緩沖區(qū)，有則合并到channel的隊列。

channel.put -> transaction.doPut：

protected void doPut(Event event) throws InterruptedException {//計算數(shù)據(jù)字節(jié)大小 int eventByteSize = (int)Math.ceil(estimateEventSize(event)/byteCapacitySlotSize); //寫入臨時緩沖區(qū)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(); ... } ... }

如果在事務(wù)期間出現(xiàn)異常，比如channel剩余空間不足，則rollback:

@Overrideprotected void doRollback() {...//拋棄數(shù)據(jù)，沒合并到channel的內(nèi)存隊列 putList.clear(); ... }

Take事務(wù)

Take事務(wù)分為以下階段：

• doTake:先將數(shù)據(jù)取到臨時緩沖區(qū)takeList
• 將數(shù)據(jù)發(fā)送到下一個節(jié)點
• doCommit:如果數(shù)據(jù)全部發(fā)送成功，則清除臨時緩沖區(qū)takeList
• doRollback:數(shù)據(jù)發(fā)送過程中如果出現(xiàn)異常，rollback將臨時緩沖區(qū)takeList中的數(shù)據(jù)歸還給channel內(nèi)存隊列。

Sink其實是由SinkRunner線程調(diào)用Sink.process方法來了處理數(shù)據(jù)的。我們從HdfsEventSink的process方法說起，Sink類都有個process方法，用來處理傳輸數(shù)據(jù)的邏輯。：

public Status process() throws EventDeliveryException {...Transaction transaction = channel.getTransaction();...//事務(wù)開始transaction.begin();...for (txnEventCount = 0; txnEventCount < batchSize; txnEventCount++) { //take數(shù)據(jù)到臨時緩沖區(qū),實際調(diào)用的是transaction.doTake Event event = channel.take(); if (event == null) { break; } ... //寫數(shù)據(jù)到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，作用是將數(shù)據(jù)放到臨時緩沖區(qū),實際調(diào)用的是transaction.doTake:

protected Event doTake() throws InterruptedException {...//從channel內(nèi)存隊列取數(shù)據(jù)synchronized(queueLock) {event = queue.poll();}...//將數(shù)據(jù)放到臨時緩沖區(qū) takeList.put(event); ... return event; }

接著，HDFS寫線程bucketWriter將take到的數(shù)據(jù)寫到HDFS,如果批數(shù)據(jù)都寫完了，則要commit了：

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

很簡單，其實就是清空takeList而已。如果bucketWriter在寫數(shù)據(jù)到HDFS的時候出現(xiàn)異常，則要rollback:

protected void doRollback() {int takes = takeList.size();//檢查內(nèi)存隊列空間大小，是否足夠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()); } ... } ... }

轉(zhuǎn)載于:https://www.cnblogs.com/whtydn/p/4384199.html

總結(jié)

以上是生活随笔為你收集整理的Flume数据传输事务分析[转]的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。