MaxCompute與OSS非結構化數據讀寫互通（及圖像處理實例）

時間  2019-12-14
標籤 maxcompute oss 構化 數據 讀寫 互通 圖像處理 實例 简体版
原文   原文鏈接

摘要： MaxCompute做爲阿里巴巴集團內部絕大多數大數據處理需求的核心計算組件，擁有強大的計算能力，隨着集團內外大數據業務的不斷擴展，新的數據使用場景也在不斷產生。在這樣的背景下，MaxCompute（ODPS）計算框架持續演化，而原來主要面對內部特殊格式數據的強大計算能力，也正在一步步的經過新增的非結構化數據處理框架，開放給不一樣的外部數據。html

0. 前言java

MaxCompute做爲阿里巴巴集團內部絕大多數大數據處理需求的核心計算組件，擁有強大的計算能力，隨着集團內外大數據業務的不斷擴展，新的數據使用場景也在不斷產生。在這樣的背景下，MaxCompute（ODPS）計算框架持續演化，而原來主要面對內部特殊格式數據的強大計算能力，也正在一步步的經過新增的非結構化數據處理框架，開放給不一樣的外部數據。 咱們相信阿里巴巴集團的這種需求，也表明着業界大數據領域的最前沿實踐和走向，具備至關的普適性。在以前咱們已經對MaxCompute 2.0新增的非結構化框架作過總體介紹，描述了在MaxCompute上如何處理存儲在OSS上面的非結構化數據，側重點在怎樣從OSS讀取各類非結構化數據並在MaxCompute上進行計算。 而一個完整數據鏈路，讀取和計算處理以後，必然也會涉及到非結構化數據的 寫出。 在這裏咱們着重介紹一下從MaxCompute往OSS輸出非結構化數據，並提供一個具體的在MaxCompute上進行圖像處理的實例， 來展現從【OSS->MaxCompute->OSS】的整個數據鏈路閉環的實現。 至於對於KV NoSQL類型數據的輸出，在對TableStore數據處理介紹 中已經有所介紹，這裏就再也不重複。git

1. 使用前提和假設github

1.1 MaxCompute 2.0 功能算法

這裏介紹的功能基於MaxCompute新一代的2.0計算框架，目前2.0計算框架已經全面上線，默認就可以使用。sql

另外本文中使用了MaxCompute 2.0新引進的一個BINARY類型，目前在使用BINARY類型時，還須要顯性設置set odps.sql.type.system.odps2=true。express

1.2 網絡連通性與訪問權限安全

另外由於MaxCompute與OSS是兩個分開的雲計算，與雲存儲服務，因此在不一樣的部署集羣上的網絡連通性有可能影響MaxCompute訪問OSS的數據的可達性。 關於OSS的節點，實例，服務地址等概念，能夠參見OSS相關介紹。 在MaxCompute公共雲服務訪問OSS存儲，推薦使用OSS私網地址（即以-internal.aliyuncs.com結尾的host地址）。網絡

此外須要指出的是，MaxCompute計算服務要訪問TableStore數據須要有一個安全的受權通道。 在這個問題上，MaxCompute結合了阿里雲的訪問控制服務（RAM）和令牌服務（STS）來實現對數據的安全反問：併發

首先須要在RAM中受權MaxCompute訪問OSS的權限。登陸RAM控制檯，建立角色AliyunODPSDefaultRole，並將策略內容設置爲：

{
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Effect": "Allow",
      "Principal": {
        "Service": [
          "odps.aliyuncs.com"
        ]
      }
    }
  ],
  "Version": "1"
}

而後編輯該角色的受權策略，將權限AliyunODPSRolePolicy受權給該角色。

若是以爲這些步驟太麻煩，還能夠登陸阿里雲帳號點擊此處完成一鍵受權。

2. MaxCompute內置的OSS數據輸出handler

2.1 建立External Table

MaxCompute非結構化數據框架但願從根本上提供MaxCompute與各類數據的聯通，這裏的「各類數據」是兩個維度上的：

1.各類存儲介質，好比OSS
2.各類數據格式， 好比文本文件，視頻，圖像，音頻，基因，氣象等格式的數據

而數據的這兩個維度的特徵，都是經過EXTERNAL TABLE的概念來引入MaxCompute的計算體系的。 與讀取OSS數據的使用方法相似，對OSS數據進行寫操做，在如上打開安全受權通道後，也是先經過CREATE EXTERNAL TABLE語句建立出一個外部表，再經過標準MaxCompute SQL的INSERT INTO/OVERWRITE等語句來實現的，這裏先用MaxCompute內置的TsvStorageHandler爲例來講明一下用法：

DROP TABLE IF EXISTS tpch_lineitem_tsv_external;

CREATE EXTERNAL TABLE IF NOT EXISTS tpch_lineitem_tsv_external
(
orderkey BIGINT,
suppkey BIGINT,
discount DOUBLE,
tax DOUBLE,
shipdate STRING,
linestatus STRING,
shipmode STRING,
comment STRING
)
STORED BY 'com.aliyun.odps.TsvStorageHandler'    ----------------------------------------- (1)
LOCATION 'oss://oss-cn-shanghai-internal.aliyuncs.com/oss-odps-test/tsv_output_folder/'；  --(2)

這個DDL語句創建了一個外部表tpch_lineitem_tsv_external，並將前面提到的兩個維度的外部數據信息關聯到這個外部表上。

1.數據存儲介質： LOCATION 將一個OSS上的地址關聯到外部表上，對這個外部表的進行讀寫操做都會反映到這個OSS地址上。
2.數據存儲格式： StorageHandler用來代表對這些數據的讀寫操做方式，這裏使用了MaxCompute內置的 com.aliyun.odps.TsvStorageHandler， 用戶可使用這個由系統自帶的實現來讀取和寫出TSV文件。 同時用戶也能夠經過MaxCompute的SDK來自定義StorageHandler, 這個將在後面的章節介紹。
其中OSS數據存儲的具體地址的URI格式爲：

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

最後還要提到的是，在上面的DDL語句中定義了外部表的Schema, 對於數據輸出而言，這表示輸出的數據格式將由這個Schema描述。 就TSV格式而言，這個schema描述比較直觀容易理解； 而在用戶自定義的輸出數據格式上，這個schema與輸出數據的聯繫則更鬆散一些，有着更大的自由度。 在後面介紹經過自定義StorageHandler/Outputer的時候會詳細展開。

2.2 經過對External Table的 INSERT 操做實現TSV文本文件的寫出

在將OSS數據經過External Table關聯上後，對OSS文件的寫出能夠對External Table作標準的SQL INSERT OVERWRITE/INSERT INTO來操做。 具體輸出數據的來源能夠有兩種

1.數據源爲MaxCompute的內部表： 也就是說能夠經過對外表INSERT操做來實現MaxCompute內部表數據到外部存儲介質的寫出。
2.數據源爲以前經過External Table引入MaxCompute計算體系的外部數據： 這能夠用來將外部數據引入MaxCompute進行計算，而後再存儲到（不一樣的）外部存儲地址，或者甚至是不一樣的外部存儲介質（好比將TableStore數據經由MaxCompute導出到OSS）。

2.2.1 從MaxCompute內部表輸出數據到OSS

這裏先來看第一種場景：假設咱們已經有一個名爲tpch_lineitem的MaxCompute內部表，其schema能夠經過

DESCRIBE tpch_lineitem;

獲得：

+------------------------------------------------------------------------------------+
| InternalTable: YES      | Size: 241483831680                                       |
+------------------------------------------------------------------------------------+
| Native Columns:                                                                    |
+------------------------------------------------------------------------------------+
| Field           | Type       | Label | Comment                                     |
+------------------------------------------------------------------------------------+
| l_orderkey      | bigint     |       |                                             |
| l_partkey       | bigint     |       |                                             |
| l_suppkey       | bigint     |       |                                             |
| l_linenumber    | bigint     |       |                                             |
| l_quantity      | double     |       |                                             |
| l_extendedprice | double     |       |                                             |
| l_discount      | double     |       |                                             |
| l_tax           | double     |       |                                             |
| l_returnflag    | string     |       |                                             |
| l_linestatus    | string     |       |                                             |
| l_shipdate      | string     |       |                                             |
| l_commitdate    | string     |       |                                             |
| l_receiptdate   | string     |       |                                             |
| l_shipinstruct  | string     |       |                                             |
| l_shipmode      | string     |       |                                             |
| l_comment       | string     |       |                                             |
+------------------------------------------------------------------------------------+

其中有16個columns。 如今咱們但願將其中的一部分數據以TSV格式導出到OSS上面。 那麼在用上述DDL建立出External Table以後，使用以下INSERT OVERWRITE操做就能夠實現：

INSERT OVERWRITE TABLE tpch_lineitem_tsv_external
SELECT l_orderkey, l_suppkey, l_discount, l_tax, l_shipdate, l_linestatus, l_shipmode, l_comment
    FROM tpch_lineitem
    WHERE l_discount = 0.07 and l_tax = 0.01;

這裏將從內部的tpch_lineitem表中，在符合l_discount = 0.07 並 l_tax = 0.01的行中選出8個列（對應tpch_lineitem_tsv_external這個外部表的schema）按照TSV的格式寫到OSS上。 在上面這個INSERT OVERWRITE操做成功完成後，就能夠看到OSS上的對應LOCATION產生了一系列文件：

osscmd ls oss://oss-odps-test/tsv_output_folder/

2017-01-14 06:48:27 39.00B Standard oss://oss-odps-test/tsv_output_folder/.odps/.meta
2017-01-14 06:48:12 4.80MB Standard oss://oss-odps-test/tsv_output_folder/.odps/20170113224724561g9m6csz7/M1_0_0-0.tsv
2017-01-14 06:48:05 4.78MB Standard oss://oss-odps-test/tsv_output_folder/.odps/20170113224724561g9m6csz7/M1_1_0-0.tsv
2017-01-14 06:47:48 4.79MB Standard oss://oss-odps-test/tsv_output_folder/.odps/20170113224724561g9m6csz7/M1_2_0-0.tsv
...

這裏能夠看到，經過上面LOCATION指定的oss-odps-test這個OSS bucket下的tsv_output_folder文件夾下產生了一個.odps文件夾，這其中將有一些.tsv文件，以及一個.meta文件。 這樣子的文件結構是MaxCompute(ODPS)往OSS上輸出所特有的：

1.經過MaxCompute對一個OSS地址，使用INSERT INTO/OVERWRITE 外部表來作寫出操做，全部的數據將在指定的LOCATION下的.odps文件夾產生；

2.其中.odps文件夾中的.meta文件爲MaxCompute額外寫出的宏數據文件，其中用於記錄當前文件夾中有效的數據。 正常狀況下，若是INSERT操做成功完成的話，能夠認爲當前文件夾的全部數據均是有效數據。 只有在有做業失敗的狀況下須要對這個宏數據進行解析。 即便是在做業中途失敗或被kill的狀況下，對於INSERT OVERWRITE操做，再跑一次成功便可。 若是對於高級用戶，必定須要解析.meta文件的話，能夠聯繫MaxCompute技術團隊。
這裏迅速看一下這些tsv文件的內容：

osscmd cat oss://oss-odps-test/tsv_output_folder/.odps/20170113232648738gam6csz7/M1_0_0-0.tsv  

4236000067      9992377 0.07    0.01    1992-11-06      F       RAIL     across the ideas nag
4236000290      3272628 0.07    0.01    1998-04-28      O       RAIL    uriously. furiously unusual dinos int
4236000386      8081402 0.07    0.01    1994-02-19      F       RAIL    its. express, iron
4236000710      3879271 0.07    0.01    1995-03-10      F       AIR     es are carefully fluffily spe
...

能夠看到確實在OSS上產生了對應的TSV數據文件。

最後，你們可能也注意到了，這個INSERT OVERWRITE操做產生了多個TSV文件，對於MaxCompute內置的TSV/CSV處理來講，產生的文件數目與對應SQL stage的併發度是相同的，在上面這個例子中，INSER OVERWITE ... SELECT ... FROM ...; 的操做在源數據表(tpch_lineitem) 上分配了1000個mapper，因此最後產生了1000個TSV文件的。 若是須要控制TSV文件的數目，能夠配合MaxCompute的各類靈活語義和配置來實現。 好比若是須要強制產生一個TSV文件，那在這個特定例子中，能夠在INSER OVERWITE ... SELECT ... FROM ...最後加上一個DISTRIBUTE BY l_discount, 就能夠在最後插入僅有一個Reducer的Reduce stage, 也就會只輸出一個TSV文件了：

osscmd ls oss://oss-odps-test/tsv_output_folder/

2017-01-14 08:03:41 39.00B Standard oss://oss-odps-test/tsv_output_folder/.odps/.meta
2017-01-14 08:03:35 4.20GB Standard oss://oss-odps-test/tsv_output_folder/.odps/20170113234037735gcm6csz7/R2_1_33_0-0.tsv

能夠看到在增長了DISTRIBUTE BY l_discount後，如今一樣的數據只了一個輸出TSV文件，固然這個文件的size就大多了。 這方面的調控技巧還有不少，都是能夠依賴SQL語言的靈活性，數據自己的特性，以及MaxCompute計算相關設置來實現的，這裏就不深刻展開了。

2.2.2 以MaxCompute爲計算介質，實現不一樣存儲介質之間的數據轉移

External Table做爲一個MaxCompute與外部存儲介質的一個切入點，以前已經介紹過對OSS數據的讀取以及TableStore數據的操做，結合對外部數據讀取和寫出的功能，就能夠實現經過External Table實現各類各樣的數據計算/存儲鏈路，好比：

1.讀取External Table A關聯的OSS數據，在MaxCompute上作複雜計算處理，並輸出到External Table B關聯的OSS地址

2.讀取External Table A關聯的TableStore數據，在MaxCompute上作複雜計算處理，並輸出到External Table B關聯的OSS地址
而這些操做與上面數據源爲MaxCompute內部表的場景， 惟一的區別只是SELECT的來源變成一個External table，而不是MaxCompute內置表。

3. 經過自定義StorageHandler來實現數據輸出

除了使用內置的StorageHandler來實如今OSS上輸出TSV/CSV等常見文本格式，MaxCompute非結構化框架提供了通用的SDK，容許用戶對外輸出自定義數據格式文件，包括圖像，音頻，視頻等等。 這種對於用戶自定義的徹底非結構化數據格式支持，也是MaxCompute從結構化/文本類數據的一個向外擴展，在這裏咱們會以一個圖像處理的例子，來走通整個【OSS->MaxCompute->OSS】數據鏈路，尤爲着重介紹對OSS輸出文件的功能。

爲了方便你們理解，這裏先提供一個在使用用戶自定義代碼的場景下，數據在MaxCompute計算平臺上的流程：

圖片描述

從上圖能夠看出，從數據的流動和處理邏輯上理解，用戶能夠簡單地把非結構化處理框架理解成在MaxCompute計算平臺兩端有機耦合的的數據導入(Ingres)以及導出(Egress)：

1.外部的（OSS）數據通過非結構化框架轉換，會使用Java用戶容易理解的InputStream類提供給自定義代碼接口。 用戶自實現Extract邏輯只須要負責對輸入的InputStream作讀取/解析/轉化/計算，最終返回MaxCompute計算平臺通用的Record格式；

2.這些Record能夠自由的參與MaxCompute的SQL邏輯運算，這一部分計算是基於MaxCompute內置的強大結構化SQL運算引擎，並可能產生新的Record

3.運算事後的Record中再傳遞給用戶自定義的Output邏輯，用戶在這裏能夠進行進一步的計算轉換，並最終將Record裏面須要輸出的信息經過系統提供OutputStream輸出，由系統負責寫到OSS。

值得指出的是，這裏面全部的步驟都是能夠由用戶根據須要來進行自由的選擇與拼接的。 好比若是用戶的輸入就是MaxCompute的內部表，那步驟1.就沒有必要了，事實上在前面的章節2中的例子，咱們就實現了將內部表直接寫成OSS上的TSV文件的流程。 同理， 若是用戶沒有輸出的需求，步驟3. 就沒有必要，好比咱們以前介紹的OSS數據的讀取。 最後，步驟2.也是能夠省略的，好比若是用戶的全部計算邏輯都是在自定義的Extract/Output中完成，沒有進行SQL邏輯運算的須要，那步驟1.是能夠直接鏈接到步驟3.的。

理解了上面這個數據變換的流程，咱們就能夠來經過一個圖像處理例子來看看怎麼具體的經過非結構化框架在MaxCompute SQL上完整的實現非結構化數據的讀取，計算以及輸出了：

3.1 範例：OSS圖像文件 -> MaxCompute計算處理 -> OSS圖像輸出

這裏咱們先提供實現這整個【OSS->MaxCompute->OSS】數據鏈路須要用到的MaxCompute SQL query，並作簡單的註解，詳細的用戶代碼實現邏輯將在後面的3.2子章節中介紹SDK接口的時候作展開解釋。

3.1.1 關聯OSS上的原始輸入圖像到External Table: images_input_external

DROP TABLE IF EXISTS images_input_external;
CREATE EXTERNAL TABLE IF NOT EXISTS images_input_external
(
name STRING,
width BIGINT,
height BIGINT,
image BINARY
)
STORED BY 'com.aliyun.odps.udf.example.image.ImageStorageHandler'   --- (1)
WITH SERDEPROPERTIES ('inputImageFormat'='BMP' , 'transformedImageFormat' = 'JPG')  --- (2)
LOCATION 'oss://oss-cn-shanghai-internal.aliyuncs.com/oss-odps-test/dev/SampleData/test_images/mixed_bmp/'  --- (3)
USING 'odps-udf-example.jar';   --- (4)

說明：

1.用戶指明使用的用戶代碼wrapper class名字是com.aliyun.odps.udf.example.image.ImageStorageHandler，這個class及其依賴的三方庫用戶經過jar提供，具體jar名字會經過下面的USING語句（見第4點）指定。

2.經過SERDEPROPERTIES來實現參數傳遞，格式爲'key'='value', 具體用法能夠參見基本功能介紹 以及下面的用戶代碼說明

3.指定輸入圖像地址，這個地址上存放了一系列不一樣分辨率的bmp圖像文件。

4.指定包含用戶JAR包，內含自定義的StorageHandler/Extractor/Outputer，以及須要的三方庫（這裏用到了Java ImageIO庫，具體見下面用戶代碼範例）。JAR包經過ADD JAR命令上傳，能夠參見基本功能介紹。

另外要說明的是這裏指定的External Table的schema就是用戶在進行Extract操做後構造的Record格式，具體怎麼構造這個Schema用戶能夠根據須要本身根據能從輸入數據中抽取到的信息定義。 在這裏咱們定義了對於輸入圖片數據，會將圖片名稱，圖片的長和寬，以及圖片的二進制bytes抽取出來放進Record（見後面的Extractor代碼說明），因此就有了上面的【STRING，BIGINT，BIGINT，BINARY】的schema。

3.1.2 關聯OSS輸出地址到External Table: images_output_external

CREATE EXTERNAL TABLE IF NOT EXISTS images_output_external
(
image_name STRING,
image_width BIGINT,
image_height BIGINT,
outimage BINARY
)
STORED BY 'com.aliyun.odps.udf.example.image.ImageStorageHandler'
LOCATION 'oss://oss-cn-shanghai-internal.aliyuncs.com/oss-odps-test/dev/output/images_output/' ---(1)
USING 'odps-udf-example.jar';

說明： 能夠看到這裏建立關聯輸出圖像文件的External Table，使用的DDL語句，與前面關聯輸入圖像時使用的DDL語句是很是相似的：只是LOCATION不同，代表圖像數據處理後將輸出到另一個地址。 另外還有一點就是這裏咱們沒有使用SERDEPROPERTIES來進行傳參，這個只是在這個場景上沒有需求，在有需求的時候能夠用一樣的方法把參數傳遞給outputer。 固然這裏兩個DDL語句如此類似，有一個緣由是由於咱們這個例子中用戶代碼中對於Extract出的Record以及輸入給Outputer的Record使用了同樣的schema, 同時這一對Extractor和Outputer都被封裝在了同一個ImageStorageHandler裏放在同一個JAR包裏。 在實際應用中，這些都是能夠根據實際需求本身調整的，由用戶本身選擇組合和打包方式。

3.1.3 從OSS讀取原始圖片數據到MaxCompute, 計算處理，並輸出圖像到OSS

在上面的3.1.1以及3.1.2子章節中的兩個DDL語句，分別實現了把輸入OSS數據，以及計劃輸出OSS數據，分別綁定到兩個LOCATION以及指定對應的用戶處理代碼，參數等設置。 然而這兩個DDL語句對系統而言，只是進行了一些宏數據的記錄操做，並不會涉及具體的數據計算操做。 在這兩個DDL語句運行成功後，運行以下SQL語句纔會引起真正的運算。 換句話說，在Fig.1中描述的整個【OSS->MaxCompute->OSS】數據讀取/計算/輸出鏈路，實際上都是經過下面一個簡單的SQL 語句完成的：

INSERT OVERWRITE TABLE images_output_external
SELECT * FROM images_input_external
WHERE width = 1024;

這看起來就是一個標準的MaxCompute SQL語句，只不過由於涉及了images_output_external和images_input_external這兩個外部表，因此真正進行的物理操做與傳統的SQL操做會有一些區別：在這個過程當中，涉及了讀寫OSS，以及經過ImageStorageHandler這個wrapper，調用自定義的Extractor，Outputer代碼來對數據進行操做。 下面就來具體看看在這個例子中的用戶自定義代碼實現了怎樣的功能，以及具體是如何實現的。

3.2 ImageStorageHandler實現

如同以前介紹過的，MaxCompute非結構化框架經過StorageHandler這個接口來描述對各類數據存儲格式的處理。 具體來講，StorageHandler做爲一個wrapper class, 讓用戶指定本身定義的Exatractor(用於數據的讀入，解析，處理等) 以及Outputer(用於數據的處理和輸出等)。 用戶自定義的StorageHandler 應該繼承 OdpsStorageHandler，實現getExtractorClass以及getOutputerClass 兩個接口。

一般做爲wrapper class, StorageHandler的實現都很簡單，好比這裏的ImageStorageHandler 就只是經過這兩個接口指定了咱們將使用ImageExtractor以及ImageOutputer:

package com.aliyun.odps.udf.example.image;

public class ImageStorageHandler extends OdpsStorageHandler {
  @Override
  public Class<!--? extends Extractor--> getExtractorClass() {
    return ImageExtractor.class;
  }

  @Override
  public Class<!--? extends Outputer--> getOutputerClass() {
    return ImageOutputer.class;
  }
}

另外要說明的是若是肯定在使用某個StorageHandler的時候，只須要用到Extractor，或者只須要用到Outputer功能，那不須要的接口則不用實現。 好比若是咱們只須要讀取OSS數據而不須要作INSERT操做，那getOutputerClass()的實現只須要扔個NotImplemented exception就能夠了，不會被調用到。

3.3 ImageExtractor實現

由於對於SDK中Extractor接口的介紹以及對用戶如何寫一個自定義的Extractor，在以前介紹的OSS數據的讀取中已經有所涉及，因此這裏就再也不對這方面作深刻的介紹。

Extractor的工做在於讀取輸入數據並進行用戶自定義處理，那麼咱們首先來看看這裏由images_input_external這個外表綁定的OSS輸入LOCATION上存放的具體數據內容：

osscmd ls oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/                                                                                                                       

2017-01-09 14:02:01 1875.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/barbara.bmp
2017-01-09 14:02:00 768.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/cameraman.bmp
2017-01-09 14:02:00 1054.74KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/fishingboat.bmp
2017-01-09 14:01:59 257.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/goldhill.bmp
2017-01-09 14:01:59 468.80KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/house.bmp
2017-01-09 14:01:59 468.80KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/jetplane.bmp
2017-01-09 14:02:01 2.32MB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/lake.bmp
2017-01-09 14:01:59 257.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/lena.bmp
2017-01-09 14:02:00 768.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/livingroom.bmp
2017-01-09 14:02:00 768.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/pirate.bmp
2017-01-09 14:02:00 768.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/walkbridge.bmp
2017-01-09 14:02:00 1054.74KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/woman_blonde.bmp
2017-01-09 14:02:00 768.05KB Standard oss://oss-odps-test/dev/SampleData/test_images/mixed_bmp/woman_darkhair.bmp

能夠看到這個LOCATION存放了一系列bmp圖像數據，分辨率從 400 x 400 到 1200 x 1200不等。 具體在這個例子中用到的ImageExtractor的詳細代碼在github上能夠找到, 這裏只作一些簡單介紹說明該Extractor作了些什麼工做：

1.從輸入的OSS地址上使用非結構化框架提供的InputStream接口讀取圖像數據，並在本地進行以下操做

對於圖像寬度小於1024的圖片，統一放大到1024 x 1024; 對於圖像寬度大於1024的圖片，跳過不進行處理
處理過的圖片，在內存中轉存成由輸入參數指定的格式（JPG)
2.把處理後在內存中的JPG數據的原始字節存入輸出的Record中的BINARY field, 同一個Record中還將存放處理後圖像的長和寬（都是1024）, 以及原始的圖像名字（這個能夠從輸入的InputStream上獲取）；
3.填充後的Record從Extract接口返回進入MaxCompute系統；
4.在這個過程當中，用戶能夠靈活的進行各類操做，好比額外的參數驗證等。

另外要說明的是，目前Record做爲MaxCompute結構化數據處理的基本單元，有一些額外的限制，好比BINARY/STRING類型都有8MB大小的限制，可是在大部分場景下這個大小應該是能知足存儲需求的。

3.4 ImageOutputer的實現

接下來咱們着重講一下ImageOutputer的實現。 首先全部的用戶輸出邏輯都必須實現Outputer接口，具體來講有以下三個：setup, output和close, 這和Extractor的setup, extract和close三個接口基本上是對稱的。

// Base outputer class, custom outputer shall extend from this class
public abstract class Outputer{

  public abstract void setup(ExecutionContext ctx, OutputStreamSet outputStreamSet, DataAttributes attributes);

  public abstract void output(Record record) throws IOException;

  public abstract void close() throws IOException;
}

這其中setup()和close()在一個outputer中只會調用一次。 用戶能夠在setup裏面作初始化準備工做，另外一般須要把setup()傳遞進來的這三個參數保存成ouputerd的class variable, 方便以後output()或者close()接口中使用。 而close()這個接口用於方便用戶代碼的掃尾工做。

一般狀況下大部分的數據處理發生在output(Record)這個接口內。 MaxCompute系統會根據當前outputer分配處理的Record數目不斷調用，也就是對每一個輸入Record系統會調用一次 output(Record)。 系統假設在一個output(Record) 調用返回的時候，用戶代碼已經消費完這個Record, 所以在當前output(Record)返回後，系統可能將這個Record所使用的內存用做它用： 因此不推薦一個Record中的信息在跨多個output()函數調用被使用，若是必定有這個需求的話，用戶必須把相關信息經過class variable等方式自行另外保存。

3.4.1 ImageOutputer.setup()

setup用於初始化整個outputer, 在這個接口上提供了整個outputer操做過程當中可能須要的參數：

  • ExecutionContext:
    用於提供一些系統信息和接口，好比讀取resource等，在ImageOutputer這個例子中咱們沒有用到這個參數；
  • OutputStreamSet:
    用戶能夠從這個類的next()接口獲取對外輸出所須要的OutputStream，具體用法咱們在下面詳細介紹；
  • DataAttributes:
    用戶經過SERDEPROPERTIES設置的key-value參數能夠經過這個類獲取，參數獲取這裏ImageOutputer例子中沒有用到，可是Extractor上的setup參數中也有這個類，在上面的ImageExtractor用到了改功能，能夠參考一下。
    同時這個類上面還提供了一些helper接口，好比方便用戶驗證schema等。

在咱們這個ImageOutputer裏，setup()的實現比較簡單：

@Override
  public void setup(ExecutionContext ctx, OutputStreamSet outputStreamSet, DataAttributes attributes) {
    this.outputStreamSet = outputStreamSet;
    this.attributes = attributes;
    this.attributes.verifySchema(new OdpsType[]{ OdpsType.STRING, OdpsType.BIGINT, OdpsType.BIGINT, OdpsType.BINARY });
  }

只是作了簡單的初始化以及對schema的驗證。

3.4.2 ImageOutputer.output(Record) 以及 OutputStreamSet的使用

在介紹具體output()接口以前，首先咱們要來看看 OutputStreamSet, 這個類有兩個接口：

public interface OutputStreamSet{

SinkOutputStream next();
SinkOutputStream next(String fileNamePostfix);

}

兩個接口都是用來獲取一個新的SinkOutputStream(一個Java OutputStream的實現，能夠按照OutputStream使用)，兩個接口惟一的區別是next()獲取的OutputStream寫出的文件名徹底由MaxCompute系統決定，而next(String fileNamePostfix)則容許用戶提供文件名的postfix。 提供這個postfix的意義是，在輸出文件具體地址和名字格式整體由MaxCompute系統決定的前提下，用戶依然能夠定製一個方便理解的postfix。 好比使用next("_boat.jpg") 獲得的OutputStream可能對應以下一個輸出文文件：

oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-0_boat.jpg

這其中尾端的"_boat.jpg"能夠幫助用戶理解輸出文件的涵義。 若是這個 OutputStream是由next()得到的話，那對應的輸出文件可能就是這樣的：

oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-0

用戶可能就須要具體讀取這個文件才能知道這個文件中具體存放了什麼內容。

前面提到output(Record)這個接口會由系統不斷調用，可是應該強調的是，並不必定在每個Record都須要調用一次OutputStreamSet.next()接口來得到一個新的OutputStream。 事實上在大多數狀況下，咱們建議在一個Outputer裏面儘量減小調用next()的次數（最好只調用一次）。 也就是說理想狀況下，一個outpuer只應該產生一個輸出文件。 好比處理TSV這種文本格式文件，假設有5000個Record對應5000行TSV數據，那麼最理想的狀況是應該把這5000行數據所有寫到一個TSV文件中。 固然用戶可能會有各類各樣不一樣的切分輸出文件的需求：好比但願每一個文件大小控制在必定範圍，或好比文件的邊界有顯著的意義等等。

具體到當前這個圖像例子，從下面的ImageOutputer代碼實現中能夠看出，這個例子中確實是處理每一個Record就調用一次next()的，由於在當前場景中，每個輸入的Record都表示一張圖片的信息（binary bytes, 圖像名字，圖像長寬），因此這裏經過屢次調用next()來輸出多個圖片文件。 可是咱們仍是須要再次強調，調用next()的次數過多可能有一些其餘弊端，好比形成碎片化小數據在OSS上的存儲等等。 尤爲在MaxCompute這種分佈式計算系統上，由於系統自己就會調度起多個outputer進行並行計算處理，若是每一個outpuer都輸出過多文件的話，最後產生的文件數目會有一個乘性效應。 回頭來看咱們這個例子中，即便在這裏，多個圖像其實也能夠經過一個OutputStream，按照tar/tar.gz的方式寫到單個文件中，這些都是在實現具體系統中用戶須要根據本身的場景, 以及處理邏輯，輸出數據類型等信息來進行優化和tradeoff的。

在理解了這些以後，如今來具體看看ImageOutputer的實現output接口實現：

@Override
  public void output(Record record) throws IOException {
    String name = record.getString(0);
    Long width = record.getBigint(1);
    Long height = record.getBigint(2);
    ByteArrayInputStream input =  new ByteArrayInputStream(record.getBytes(3));
    BufferedImage sobelEdgeImage = getEdgeImage(input);
    OutputStream outputStream = this.outputStreamSet.next(name + "_" + width + "x" + height + "." + outputFormat);
    ImageIO.write(sobelEdgeImage, this.outputFormat, outputStream);
  }

能夠看到這裏主要就作了三件事情：

1. 根據以前保存的圖像名字，長寬信息，和編碼方式(".jpg")拼出一個帶擴展名的輸出文件名postfix。
2. 讀取圖像binary bytes，並用getEdgeImage()來利用sobel算子對圖像作邊緣檢測。

具體getEdgeImage()的實現這裏就不進行深刻解釋了： 使用了標準的sobel模板卷積算法，
有興趣看ImageOutputer源碼便可。

3. 對每個圖像產生一個新的OutputStream並將數據寫出，至此當前Record處理完畢，寫出一張圖片到OSS，output()函數返回。

3.4.3 ImageOutputer.close()

在這個例子中，outputer.close()接口沒有包含具體的實現邏輯，是個no-op。

至此咱們就介紹完了一個output的實現，如今能夠看看在運行完這個SQL query，對應OSS地址的數據：

osscmd ls oss://oss-odps-test/dev/output/images_output/ 

2017-01-15 14:36:50 215.19KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-0-barbara_1024x1024.jpg
2017-01-15 14:36:50 108.90KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-1-cameraman_1024x1024.jpg
2017-01-15 14:36:50 169.54KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-2-fishingboat_1024x1024.jpg
2017-01-15 14:36:50 214.94KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-3-goldhill_1024x1024.jpg
2017-01-15 14:36:50 71.00KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-4-house_1024x1024.jpg
2017-01-15 14:36:50 126.50KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-5-jetplane_1024x1024.jpg
2017-01-15 14:36:50 169.63KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-6-lake_1024x1024.jpg
2017-01-15 14:36:50 194.18KB Standard oss://oss-odps-test/dev/output/images_output/.odps/20170115148446219dicjab270/M1_0_-1--1-7-lena_1024x1024.jpg
...

能夠看到圖像數據按照期待格式寫到了指定地址，這裏咱們就選一個輸入圖像(lena.bmp)以及對應的輸出圖像(M1_0_-1--1-7-lena_1024x1024.jpg)看一下對比：

圖片描述

這個例子中整個圖像處理流程已經經過如上的SQL query完成。 而從上面展現的ImageExtractor以及ImageOutputer 源代碼，咱們能夠看出整個過程當中用戶的邏輯基本與寫單機圖像處理程序無異，用戶的代碼只須要在Extractor上作InputStream到Record的準換，而在Outputer上作反向的Record到OutputSteam的寫出處理，其餘核心的處理邏輯實現基本和單機算法實現相同，在用戶的層面，並不用去操心底層分佈式系統的細節以及MaxCompute和OSS的交互。

3.5 數據處理步驟的靈活性

從上面這個例子中咱們也能夠看出，在一個完整的【OSS->MaxCompute->OSS】數據流程中，Extractor和Outputer中涉及的具體計算邏輯其實也並不必定會有一個很是明確的邊界。 Extractor和Outputer只要各自完成所需的轉換Record/Stream的轉換，具體的額外算法邏輯在兩個地方都有機會完成。 好比上面這個例子的整個流程涉及了以下圖像處理相關的運算：

1.圖像的縮放 （統一到 1024 x 1024）
2.圖像格式的轉換 (BMP -> JPG)
3.圖像的Sobel邊緣檢測

上面的例子實現中，把1. 和 2. 放在ImageExtractor中完成，而3.則放在ImageOutputer中完成，但並非惟一的選擇。 咱們徹底能夠把全部3個步驟都放在ImageExtractor中完成，讓ImageOutputer只作Record到寫出最後圖像的操做；也能夠在ImageExtractor中只作讀取原始binary到Recrod, 而把全部3個圖像處理步驟都放在ImageOutputer中進行，等等。 具體進行怎樣的選擇，用戶能夠徹底根據須要本身實現。

另一個系統設計的點是若是對於一個數據須要作重複的運算，那能夠考慮將數據從OSS中經過Extractor讀出進MaxCompute，而後存儲成MaxCompute的內置表格再進行（屢次）的計算。 這個對於MaxCompute和OSS沒有進行混布，不在一個物理網絡上的場景尤爲有意義： MaxCompute從內置表中讀取數據無疑要比從外部OSS存儲服務中讀出數據要有效得多。 在上面3.1.3子章節中的圖像處理例子，這個INSER OVERWITE操做：

INSERT OVERWRITE TABLE images_output_external
SELECT * FROM images_input_external
WHERE width = 1024;

就能夠改寫成兩個分開的語句：

INSERT OVERWRITE TABLE images_internal
SELECT * FROM images_input_external
WHERE width = 1024;

INSERT OVERWRITE TABLE images_output_external
SELECT * FROM image_internal;

經過把數據寫到一個內部images_internal表中，後面若是有屢次讀取數據的需求的話，就能夠再也不去訪問外部OSS了。 這裏也能夠看到MaxCompute非結構化框架以及SQL語法自己提供了很是高的靈活性和可擴展性，用戶能夠根據實際計算的不一樣模式/場景/需求，來在上面完成各類各樣的數據計算工做流。

  1. 結語

非結構化數據處理框架隨着MaxCompute 2.0一塊兒推出，意在豐富MaxCompute平臺的數據處理生態，來打通阿里雲核心計算平臺與阿里雲各個重要存儲服務之間的數據鏈路。 在以前介紹過的讀取OSS以及處理TableStore數據的總體方案後，本文側重介紹數據往OSS的輸出方案，並依託一個圖像處理的處理實例，展現了【OSS->MaxCompute->OSS】整個數據鏈路的實現。 在這些新功能的基礎上，咱們但願實現整個阿里雲計算與數據的生態融合： 在不一樣的項目上，咱們已經看到了在MaxCompute上處理OSS上的海量視頻，圖像等非結構化數據的巨大潛力。 從此隨着這個生態的豐富，咱們指望OSS數據，TableStore數據以及MaxCompute內部存儲的數據，都能在MaxCompute的核心計算引擎上進行融合，從而產生更大的價值。

原文連接

閱讀更多幹貨好文，請關注掃描如下二維碼：
圖片描述

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程