Flume資料傳輸事務分析

本文基於thriftsource,memorychannel,hdfssink三個元件，對flume資料傳輸的事務進行分析，如果使用的是其他元件，flume事務具體的處理方式將會不同。一般情況下，用memorychannel就好了，我們公司用的就是這個，filechannel速度慢，雖然提供日誌級別的資料恢復，但是一般情況下，不斷電memorychannel是不會丟資料的。

flume提供事物操作，保證使用者的資料的可靠性，主要體現在：

flume在對channel進行put和take操作的時候，必須要用事物包住,比如：

channel ch = new memorychannel();
transaction txn = ch.gettransaction();
//事物開始
txn.begin();
try  catch (throwable t) 
} finally

put事務可以分為以下階段：

我們從source資料接收到寫入channel這個過程對put事物進行分析。

thriftsource會spawn多個worker執行緒(thriftsourcehandler)去處理資料，worker處理資料的介面，我們只看batch批量處理這個介面：

@override
listflumeevents = lists.newarraylist();
for(thriftflumeevent event : events) 
//channelprocessor,在source初始化的時候傳進來.將資料寫入對應的channel
getchannelprocessor().processeventbatch(flumeevents);
...return status.ok;
}

事務邏輯都在processeventbatch這個方法裡：

public
void
processeventbatch(listevents) 
//提交，將資料寫入channel的佇列中
tx.commit();
} catch (throwable t) 
}...
}

每個worker執行緒都擁有乙個transaction例項，儲存在channel(basicchannelsemantics)裡的threadlocal變數currenttransaction.

那麼，事務到底做了什麼？

實際上，transaction例項包含兩個雙向阻塞佇列linkedblockingdeque(感覺沒必要用雙向佇列，每個執行緒寫自己的putlist，又不是多個執行緒？),分別為：

對於put事物操作，當然是只用到putlist了。putlist就是乙個臨時的緩衝區，資料會先put到putlist,最後由commit方法會檢查channel是否有足夠的緩衝區，有則合併到channel的佇列。

channel.put -> transaction.doput：

protected
void
doput(event event) throws interruptedexception 
putbytecounter += eventbytesize;
}

transaction.commit：

@override
protected
void
docommit() throws interruptedexception }}
//清除臨時佇列
putlist.clear();
...}
...}

如果在事務期間出現異常，比如channel剩餘空間不足，則rollback:

@override
protected
void
dorollback()

take事務分為以下階段：

sink其實是由sinkrunner執行緒呼叫sink.process方法來了處理資料的。我們從hdfseventsink的process方法說起，sink類都有個process方法，用來處理傳輸資料的邏輯。：

public status process() throws eventdeliveryexception 
...//寫資料到hdfs
...// flush all pending buckets before committing the transaction
for (bucketwriter bucketwriter : writers) 
//commit
transaction.commit();
...} catch (ioexception eio)  finally 
}

大致流程圖：

接著看看channel.take，作用是將資料放到臨時緩衝區,實際呼叫的是transaction.dotake:

protected event dotake() throws interruptedexception 
...//將資料放到臨時緩衝區
takelist.put(event);
...return
event;
}

接著，hdfs寫執行緒bucketwriter將take到的資料寫到hdfs,如果批資料都寫完了，則要commit了：

protected
void
docommit() throws interruptedexception

很簡單，其實就是清空takelist而已。如果bucketwriter在寫資料到hdfs的時候出現異常，則要rollback:

protected
void
dorollback() 
...}
...}

Flume資料傳輸事務分析

C SFTP資料傳輸

資料傳輸方式

flex wcf 資料傳輸

Flume資料傳輸事務分析

C SFTP資料傳輸

資料傳輸方式

flex wcf 資料傳輸

相關推薦