摘要：會盡最大可能容錯，保證數(shù)據(jù)上傳上去。多線程上傳示例多線程上傳時，每個線程只需要打開一個往里面寫數(shù)據(jù)就行了。多個進(jìn)程共享由于一個的上傳狀態(tài)是通過維護(hù)一個實現(xiàn)的，對于多線程程序來講，通過鎖很容易實現(xiàn)資源的分配。

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

本文用到的

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

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

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

High Level API

MaxCompute Java SDK 在 0.21.3-public 之后新增了 BufferredWriter 這個更高層的 API，簡化了數(shù)據(jù)上傳的過程，并且提供了容錯的功能。 BufferedWriter 對用戶隱藏了 block 這個概念，從用戶角度看，就是在 session 上打開一個 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();

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

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

多線程上傳示例

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

class UploadThread extends Thread {
  private UploadSession session;
  private static int RECORD_COUNT = 1200;

  public UploadThread(UploadSession session) {
    this.session = session;
  }

  @Override
  public 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();
 }

更多控制

重試策略

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

用戶可以選擇三種重試回避策略：指數(shù)回避（EXPONENTIAL_BACKOFF）、線性時間回避（LINEAR_BACKOFF）、常數(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)格的要求，可以通過下面這個接口修改緩沖區(qū)占內(nèi)存的字節(jié)數(shù)（bytes）：

writer.setBufferSize(1024*1024);

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

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

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

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

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

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

Notes

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

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

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

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

原文鏈接

閱讀更多干貨好文，請關(guān)注掃描以下二維碼：