為什么80%的碼農(nóng)都做不了架構(gòu)師？>>> ??

摘要：?MaxCompute 的數(shù)據(jù)上傳接口（Tunnel）定義了數(shù)據(jù) block 的概念：一個(gè) block 對(duì)應(yīng)一個(gè) http request，多個(gè) block 的上傳可以并發(fā)而且是原子的，一次同步請(qǐng)求要么成功要么失敗，不會(huì)污染其他的 block。這種設(shè)計(jì)對(duì)于服務(wù)端來講十分簡潔，但是也把記錄狀態(tài)做 fa.

本文用到的

阿里云數(shù)加－大數(shù)據(jù)計(jì)算服務(wù)MaxCompute產(chǎn)品地址：https://www.aliyun.com/product/odps

MaxCompute 的數(shù)據(jù)上傳接口（Tunnel）定義了數(shù)據(jù) block 的概念：一個(gè) block 對(duì)應(yīng)一個(gè) http request，多個(gè) block 的上傳可以并發(fā)而且是原子的，一次同步請(qǐng)求要么成功要么失敗，不會(huì)污染其他的 block。這種設(shè)計(jì)對(duì)于服務(wù)端來講十分簡潔，但是也把記錄狀態(tài)做 failover 的工作交給了客戶端。

用戶在使用 Tunnel SDK 編程時(shí)，需要對(duì) block 這一層的語義進(jìn)行認(rèn)知，并且驅(qū)動(dòng)數(shù)據(jù)上傳的整個(gè)過程[1]，并且自己進(jìn)行容錯(cuò)，畢竟『網(wǎng)絡(luò)錯(cuò)誤是正常而不是異常』。由于用戶文檔中并沒有強(qiáng)調(diào)這一點(diǎn)的重要性，導(dǎo)致很多用戶踩了坑，一種常見的出錯(cuò)場景是，當(dāng)客戶端寫數(shù)據(jù)的速度過慢，兩次 write 的間隔超時(shí)[2]，導(dǎo)致整個(gè) block 上傳失敗。

High Level API

MaxCompute Java SDK 在 0.21.3-public ?之后新增了 BufferredWriter 這個(gè)更高層的 API，簡化了數(shù)據(jù)上傳的過程，并且提供了容錯(cuò)的功能。 BufferedWriter 對(duì)用戶隱藏了 block 這個(gè)概念，從用戶角度看，就是在 session 上打開一個(gè) writer 然后往里面寫記錄即可：

RecordWriter writer = null;try {int i = 0; writer = uploadSession.openBufferedWriter();Record product = uploadSession.newRecord();for (String item : items) {product.setString("name", item);product.setBigint("id", i);writer.write(product);i += 1;} } finally {if (writer != null) {writer.close();} } uploadSession.commit();

具體實(shí)現(xiàn)時(shí) BufferedWriter 先將記錄緩存在客戶端的緩沖區(qū)中，并在緩沖區(qū)填滿之后打開一個(gè) http 連接進(jìn)行上傳。BufferedWriter 會(huì)盡最大可能容錯(cuò)，保證數(shù)據(jù)上傳上去。

• 由于屏蔽了底層細(xì)節(jié)，這個(gè)接口可能并不適合數(shù)據(jù)預(yù)劃分、斷點(diǎn)續(xù)傳、分批次上傳等需要細(xì)粒度控制的場景。

多線程上傳示例

多線程上傳時(shí)，每個(gè)線程只需要打開一個(gè) writer 往里面寫數(shù)據(jù)就行了。

class UploadThread extends Thread {private UploadSession session;private static int RECORD_COUNT = 1200;public UploadThread(UploadSession session) {this.session = session;}@Overridepublic void run() {RecordWriter writer = up.openBufferedWriter();Record r = up.newRecord();for (int i = 0; i < RECORD_COUNT; i++) {r.setBigint(0, i);writer.write(r);}writer.close();} };public class Example {public static void main(String args[]) {// 初始化 MaxCompute 和 tunnel 的代碼TableTunnel.UploadSession uploadSession = tunnel.createUploadSession(projectName, tableName);UploadThread t1 = new UploadThread(up);UploadThread t2 = new UploadThread(up);t1.start();t2.start();t1.join();t2.join();uploadSession.commit();}

更多控制

重試策略

由于底層在上傳出錯(cuò)時(shí)會(huì)回避一段固定的時(shí)間并進(jìn)行重試，但如果你的程序不想花太多時(shí)間在重試上，或者你的程序位于一個(gè)極其惡劣的網(wǎng)絡(luò)環(huán)境中，為此 TunnelBufferedWriter 允許用戶配置重試策略。

用戶可以選擇三種重試回避策略：指數(shù)回避（EXPONENTIAL_BACKOFF）、線性時(shí)間回避（LINEAR_BACKOFF）、常數(shù)時(shí)間回避（CONSTANT_BACKOFF）。

例如下面這段代碼可以將，write 的重試次數(shù)調(diào)整為 6，每一次重試之前先分別回避 4s、8s、16s、32s、64s 和 128s（從 4 開始的指數(shù)遞增的序列）。

RetryStrategy retry = new RetryStrategy(6, 4, RetryStrategy.BackoffStrategy.EXPONENTIAL_BACKOFF)writer = (TunnelBufferedWriter) uploadSession.openBufferedWriter(); writer.setRetryStrategy(retry);

緩沖區(qū)控制

如果你的程序?qū)?JVM 的內(nèi)存有嚴(yán)格的要求，可以通過下面這個(gè)接口修改緩沖區(qū)占內(nèi)存的字節(jié)數(shù)（bytes）：

writer.setBufferSize(1024*1024);

默認(rèn)配置每一個(gè) Writer 的 BufferSize 是 10 MiB。TunnelBufferedWriter 一次 flush buffer 的操作上傳一個(gè) block 的數(shù)據(jù)[3]。

多個(gè)進(jìn)程共享 Session

由于一個(gè) Session 的上傳狀態(tài)是通過維護(hù)一個(gè) block list 實(shí)現(xiàn)的，對(duì)于多線程程序來講，通過鎖很容易實(shí)現(xiàn)資源的分配。但對(duì)于兩個(gè)進(jìn)程空間里的程序想要復(fù)用一個(gè) Session 時(shí)，必須通過一種機(jī)制對(duì)資源進(jìn)行隔離。

具體地，在 getUploadSession 的時(shí)候，必須指定這個(gè)共享這個(gè) Session 的進(jìn)程數(shù)目，以及一個(gè)用來區(qū)分進(jìn)程的 global id：

//程序1：這個(gè) session 將被兩個(gè) writer 共享，我是其中第 0 個(gè) TableTunnel.UploadSession up = tunnel.getUploadSession(projectName, tableName, sid, 2, 0); writer = session.openBufferedWriter();//程序1：這個(gè) session 將被兩個(gè) writer 共享，我是其中第 1 個(gè) TableTunnel.UploadSession up = tunnel.getUploadSession(projectName, tableName, sid, 2, 1); writer = session.openBufferedWriter();

Notes

[1] 一次完整的上傳流程通常包括以下步驟：

先對(duì)數(shù)據(jù)進(jìn)行劃分
為每個(gè)數(shù)據(jù)塊指定 block id，即調(diào)用 openRecordWriter(id)
然后用一個(gè)或多個(gè)線程分別將這些 block 上傳上去
并在某個(gè) block 上傳失敗以后，需要對(duì)整個(gè) block 進(jìn)行重傳
在所有 block 都上傳以后，向服務(wù)端提供上傳成功的 blockid list 進(jìn)行校驗(yàn)，即調(diào)用 session.commit([1,2,3,...])
[2] 因?yàn)槭褂瞄L連接，服務(wù)端有計(jì)時(shí)器判斷是否客戶端是否 alive

[3] block 在服務(wù)端有 20000 個(gè)的數(shù)量上限，如果 BufferSize 設(shè)得太小會(huì)導(dǎo)致 20000 個(gè) block 很快被用光

[4] Session的有效期為24小時(shí)，超過24小時(shí)會(huì)導(dǎo)致數(shù)據(jù)上傳失敗

原文鏈接

轉(zhuǎn)載于:https://my.oschina.net/yunqi/blog/1785569

總結(jié)

以上是生活随笔為你收集整理的MaxCompute Tunnel SDK数据上传利器——BufferedWriter使用指南的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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