MaxCompute(ODPS)上處理非結構化數據的Best Practice

摘要： 隨着MaxCompute（ODPS）2.0的上線，新增的非結構化數據處理框架也推出一系列的介紹文章，包括 MaxCompute上如何訪問OSS數據, 基本功能用法和總體介紹，側重介紹讀取OSS數據進行計算處理； 本文：MaxCompute(ODPS)上處理非結構化數據的Best Practice。

隨着MaxCompute（ODPS）2.0的上線，新增的非結構化數據處理框架也推出一系列的介紹文章，包括

一、MaxCompute上如何訪問OSS數據, 基本功能用法和總體介紹，側重介紹讀取OSS數據進行計算處理；

二、MaxCompute上處理非結構化數據的Best Practice。 基於非結構化框架實現原理，提供一些最佳實踐總結；

三、MaxCompute訪問TableStore(OTS) 數據, 着重介紹經過非結構化框架來訪問計算KV（TableStore/OTS）數據；

四、MaxCompute到OSS的非結構化數據輸出（及圖像處理實例）：介紹了非結構化輸出功能，並經過圖像處理等範例，說明怎樣經過MaxCompute的計算能力，打通整個OSS -> MaxCompute -> OSS的數據處理閉環；

五、如何在MaxCompute上處理存儲在OSS上的開源格式數據, 介紹對於存儲在OSS上的常見開源數據（ORC, PARQUET, AVRO等）格式，如何經過非結構化框架進行處理。

本文是這系列中的第【2】篇。

 前言

隨着MaxCompute（原ODPS）非結構化數據處理框架的推出，在SQL線上打通了MaxCompute與OSS數據之間的計算數據鏈接生態，咱們看到了視頻，圖像，音頻以及基因，氣象等各類各類各樣數據在MaxCompute平臺上實現了與傳統結構化數據的無縫融合。以前咱們提供了在MaxCompute非結構化框架處理OSS上數據的總體介紹，在基本功能實現後，咱們收到用戶許多關於優化和怎樣最好的使用非結構化功能的問題。 這裏經過分析非結構化框架底層的一些實現原理以及咱們看到的一些使用場景，提供一些關於Best Practice的總結，方便你們更有效的在MaxCompute中處理各類數據。

1. 數據在OSS上的存儲

1.1 OSS LOCATION 的選擇

MaxCompute經過在EXTERNAL TABLE上的LOCATION cluase來指定須要處理的OSS數據地址【注：本文假設用戶對於非結構化框架，包括EXTERNABLE TABLE， StorageHanlder等的定義等都有比較好的瞭解，相關細節這裏再也不具體說明。 有疑問能夠先參考以前的基本功能介紹】。其中LOCATION將指向一個OSS的一個目錄（或者更準確的說，是一個以‘/’結尾的地址），其中LOCATION爲標準URI格式：

LOCATION 'oss://${endpoint}/${bucket}/${userPath}/'

 對於數據安全比較敏感的場景，好比在多用戶場景或者公共雲上，則推薦採用上述方式，再也不LOCATION上使用AK，而是經過STS/RAM體系事先進行鑑權（參見基本功能介紹）。

LOCATION的選擇有幾點要注意：

• 不容許使用oss的root bucket做爲LOCATION, 也就是說${userPath}不能夠爲空，這個要求源自OSS對root bucket下存放內容的一些限制。
• LOCATION不能指向一個單獨文件，也就是說，相似oss://oss-cn-hangzhou.aliyuncs.com/mybucket/directory/data.csv 這種LOCATION是無效的。 若是隻有一個文件要處理，則應該提供該文件的父目錄。

1.2 數據文件的存儲和處理：小文件和大文件

在分佈式計算系統中，文件的大小對於整個系統的運行效率，性能等都有比較大的相關性。 這裏對MaxCompute對非結構化數據的相關處理機制作一個介紹，並分析幾種有表明性的場景（e.g., 小文件和大文件），總結了幾個針對MaxCompute計算場景中，比較好的OSS文件存儲建議。

• 小文件：一般小文件每每伴隨着超大的文件數目，這對於分佈式計算系統來講，有兩個問題：

  1. 大的文件數，會致使在進行文件分片時, 獲取文件宏信息的overhead較大，致使planning和分片比較耗時，好比一個100萬個文件的oss LOCATION, planning的耗時可能在分鐘以上的量級。
  2. 打開每一個OSS文件是有ovehead的，碎片化的小文件會帶來額外的讀取開銷。 好比從OSS讀取1000個10KB大小的文件，相比讀取一個10MB的的文件，耗時可能在10倍以上。 對大量小文件的訪問將帶來整個分佈式系統更多的網絡開銷，下降實際上有效的IO throughput。

  因此整體上不推薦在一個OSS目錄中存放過多的文件。 能夠從另外一個方面，考慮將Externable Table作partition，儘可能在partition的子粒度上進行數據處理。 另外，在適用的場景下，能夠考慮使用tar文件，好比把多個圖像文件打在一個tar文件中再保存到OSS上面。 若是是文本文件，MaxCompute的built-in StorageHandler (好比com.aliyun.odps.CsvStorageHandler或者com.aliyun.odps.TsvStorageHandler) 是能自動從tar文件中讀取數據的。 若是用戶本身定義的StorageHandler/Extractor，也能夠在用戶代碼中使用Java中的tar處理類，好比直接使用Apache common 的TarArchiveInputStream來訪問。

• 大文件：與小文件相對的，是另一個極端： 超大文件。 分佈式系統的精髓是分而治之的思想：對數據進行分片，經過併發處理多個分片來加快海量數據的處理。 在極限狀況下，若是海量數據存在一個沒法被切割處理的單個文件中，那併發度就被降成爲1，這樣子的「分佈式系統」就失去了意義。 即便沒有那麼極端，多個超大文件（好比每一個幾十GB），對分佈式系統也是不友好的：大的文件處理可能須要單獨佔用大量系統資源，給資源調度帶來困難，另外還容易形成長尾，失敗重跑代價太高等問題。 因此從MaxCompute處理計算的角度，也不推薦在OSS上使用超大文件保存數據。

總結一下， 做爲一個總體上的指導原則，MaxCompute非結構框架推薦以下比較理想的OSS數據存儲方案：

1. 數據文件根據應用特性，分文件夾存儲，不推薦一個文件夾中存儲10萬以上個文件。 能夠考慮使用tar打包多個文件來做爲下降物理文件數目的方法。

2. 比較適中的文件大小以及均勻分佈的數據文件，能更合理的使用各類系統資源， 從而提升分佈式處理效率。 對MaxCompute非結構化框架而言，單個文件大小在1MB-2GB是比較理想的狀況。

1.3 MaxCompute訪問OSS的網絡連通以及速度

MaxComput和OSS做爲獨立的分佈式計算和存儲服務，在不一樣的部署集羣上的網絡連通性有可能影響MaxCompute訪問OSS的數據的可達性。 網絡的連通性總體服從七網隔離的原則，具體一點來講有幾點：

1. MaxCompute的公共雲集羣上的計算應該訪問OSS的外部集羣，另外推薦須要訪問的OSS集羣與MaxCompute計算集羣在物理上儘可能靠近。關於OSS公共雲上的訪問域名以及對應數據中心能夠參考OSS文檔。

在MaxCompute併發訪問OSS的狀況下，一個須要特別注意的是OSS具備限流機制，默認狀況下一個OSS帳號的訪問流量是限制在5Gb/s，也就是600MB/s左右。 在MaxComput的高併發度下（好比1000個以上的計算節點），OSS數據下載的速度可能將再也不受限於單機網絡速度，而取決與OSS的整體流量限速。 在這種狀況下，徹底可能出現單個計算節點的下載速度低於1MB/s。 固然OSS的限流是能夠特別配置的，若是有超大量的數據計算需求，能夠聯繫OSS團隊調高對應帳戶的具體的限流上限。

2. 在用戶自定義StorageHandler/Extractor中對輸入數據的處理

除了提供幾個內置的StorageHandler用來處理CSV, TSV以及Apache ORC文件之外，MaxCompute同時開發了非結構化Java SDK來方便用戶對數據進行解析和處理。 經過這樣的方法，擴展整個非結構化數據處理的生態，對接視頻，圖像，音頻，基因，氣象等數據處理的能力。 簡單的來講， MaxCompute封裝了分佈式系統的細節，使用Java InputStream 的一個加強子類來將作輸入數據與用戶代碼的對接。 這樣的接口設計區別於Hive的SerDe， RowFormatter等多層封裝，提供了更天然的徹底非結構化數據入口， 用戶能得到原始數據流，用相似單機程序類似的邏輯進行處理。 固然，基於分佈式系統的處理原則，仍是有一些Best Practice推薦用戶遵照。

2.1 輸入數據流的處理模式

對於輸入數據流（InputStream），推薦在獲取數據bytes後能直接在內存中直接處理。 最理想的狀況是，能針對輸入數據作流式的「邊讀邊計算」的處理。 固然，對於某些數據格式，因爲數據自己的特性，很難作到徹底的流式處理：好比對於某些圖片/音頻數據格式，一張文件必須徹底讀入才能得到正確的編碼信息以及其餘特性，那這種狀況下，在文件自己不是很大的狀況下，能夠把文件徹底讀入本地內存，再行處理。 效率比較低的一種方式是把數據文件下載到本地，而後再經過FileStream讀取本地文件進行處理，這樣的處理模式有兩個問題：

1. 做爲分佈式系統，爲了實現資源隔離和保護計算節點的健康度，通常不推薦往本地磁盤寫文件（尤爲是大文件）。在MaxCompue計算系統上，用戶的Java代碼對本地文件近些讀寫操做須要另外申請權限，或者打開隔離選項（整體計算性能會降低）。
2. 數據寫入到本地落盤，再讀取，性能上有額外的損耗。
3. 對於比較大的數據（好比10GB或更大的文件），運算節點的磁盤空間沒法作保證，存在磁盤被寫爆的可能

2.2 三方庫使用

在非結構化數據的處理線上，常常遇到的一個需求是把單機的數據處理機制，經過MaxCompute非結構化數據框架，遷移到分佈式系統上執行。 好比但願同過ffmpeg來直接讀取視頻數據，或者但願經過Netcdf-Java來直接處理氣象的netcdf/grib格式數據。 而這些三方庫每每有一些共同的特性/侷限性，好比

• 多是基於C/C++，因此須要經過JNI來運行native代碼
• 多是面對單機實現，因此數據的入口常常是一個本地的文件地址

在這些狀況下，非結構化框架均有對應的方式來支持。 好比在隔離打開的狀況下容許JNI的使用，以及經過權限審批容許數據下載到本機臨時文件等等。 從長期來說，MaxCompute框架自己也認同使用native C/C++代碼庫，來處理各類特定的數據格式，將是沒法避免的，因此會從框架自己安全等方面來解決這個問題，可是對於讀取數據到本地再作處理，從本質上是一種比較大的額外消耗，仍是推薦經過直接處理輸入數據的方式來作，好比改動NETCDF-JAVA的實現，把輸入接口經過FilePath->FileStream改爲直接使用InputStream等。

3. 結語

MaxCompute非結構化框架是隨着MaxCompute2.0推出的新功能，除了處理OSS上面的非結構化數據以外，最近也打通了與TableStore(OTS)的數據鏈路。 框架自己也還在不斷的發展和完善，包括和MaxCompute優化器以及和整個UDF框架更緊密的結合和擴展等等。 在這裏先從現有系統的實現和咱們收到的一些反饋，總結提煉了一些處理非結構化數據的最佳實踐，也但願獲得更多的反饋，把框架功能作到更優。 後繼咱們也會結合具體的使用場景，好比城市大腦上的離線視頻圖像處理等，來提供一些更具體的使用範例。

原文連接