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

摘要：?MaxCompute作為阿里巴巴集團內(nèi)部絕大多數(shù)大數(shù)據(jù)處理需求的核心計算組件，擁有強大的計算能力，隨著集團內(nèi)外大數(shù)據(jù)業(yè)務(wù)的不斷擴展，新的數(shù)據(jù)使用場景也在不斷產(chǎn)生。在這樣的背景下，MaxCompute（ODPS）計算框架持續(xù)演化，而原來主要面對內(nèi)部特殊格式數(shù)據(jù)的強大計算能力，也正在一步步的通過新增的非結(jié)構(gòu)化數(shù)據(jù)處理框架，開放給不同的外部數(shù)據(jù)。

?前言

MaxCompute作為阿里巴巴集團內(nèi)部絕大多數(shù)大數(shù)據(jù)處理需求的核心計算組件，擁有強大的計算能力，隨著集團內(nèi)外大數(shù)據(jù)業(yè)務(wù)的不斷擴展，新的數(shù)據(jù)使用場景也在不斷產(chǎn)生。在這樣的背景下，MaxCompute（ODPS）計算框架持續(xù)演化，而原來主要面對內(nèi)部特殊格式數(shù)據(jù)的強大計算能力，也正在一步步的通過新增的非結(jié)構(gòu)化數(shù)據(jù)處理框架，開放給不同的外部數(shù)據(jù)。 我們相信阿里巴巴集團的這種需求，也代表著業(yè)界大數(shù)據(jù)領(lǐng)域的最前沿實踐和走向，具有相當?shù)钠者m性。在之前我們已經(jīng)對MaxCompute 2.0新增的非結(jié)構(gòu)化框架做過整體介紹，描述了在MaxCompute上如何處理存儲在OSS上面的非結(jié)構(gòu)化數(shù)據(jù)，側(cè)重點在怎樣從OSS讀取各種非結(jié)構(gòu)化數(shù)據(jù)并在MaxCompute上進行計算。 而一個完整數(shù)據(jù)鏈路，讀取和計算處理之后，必然也會涉及到非結(jié)構(gòu)化數(shù)據(jù)的?寫出。 在這里我們著重介紹一下從MaxCompute往OSS輸出非結(jié)構(gòu)化數(shù)據(jù)，并提供一個具體的在MaxCompute上進行圖像處理的實例，?來展示從【OSS->MaxCompute->OSS】的整個數(shù)據(jù)鏈路閉環(huán)的實現(xiàn)。 至于對于KV NoSQL類型數(shù)據(jù)的輸出，在對TableStore數(shù)據(jù)處理介紹?中已經(jīng)有所介紹，這里就不再重復。

1. 使用前提和假設(shè)

1.1 MaxCompute 2.0 功能

這里介紹的功能基于MaxCompute新一代的2.0計算框架，目前2.0計算框架已經(jīng)全面上線，默認就可使用。

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

1.2 網(wǎng)絡(luò)連通性與訪問權(quán)限

另外因為MaxCompute與OSS是兩個分開的云計算，與云存儲服務(wù)，所以在不同的部署集群上的網(wǎng)絡(luò)連通性有可能影響MaxCompute訪問OSS的數(shù)據(jù)的可達性。 關(guān)于OSS的節(jié)點，實例，服務(wù)地址等概念，可以參見OSS相關(guān)介紹。 在MaxCompute公共云服務(wù)訪問OSS存儲，推薦使用OSS私網(wǎng)地址（即以-internal.aliyuncs.com結(jié)尾的host地址）。

此外需要指出的是，MaxCompute計算服務(wù)要訪問TableStore數(shù)據(jù)需要有一個安全的授權(quán)通道。 在這個問題上，MaxCompute結(jié)合了阿里云的訪問控制服務(wù)（RAM）和令牌服務(wù)（STS）來實現(xiàn)對數(shù)據(jù)的安全反問：

首先需要在RAM中授權(quán)MaxCompute訪問OSS的權(quán)限。登錄RAM控制臺，創(chuàng)建角色AliyunODPSDefaultRole，并將策略內(nèi)容設(shè)置為：

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

然后編輯該角色的授權(quán)策略，將權(quán)限AliyunODPSRolePolicy授權(quán)給該角色。

如果覺得這些步驟太麻煩，還可以登錄阿里云賬號點擊此處完成一鍵授權(quán)。

2. MaxCompute內(nèi)置的OSS數(shù)據(jù)輸出handler

2.1 創(chuàng)建External Table

MaxCompute非結(jié)構(gòu)化數(shù)據(jù)框架希望從根本上提供MaxCompute與各種數(shù)據(jù)的聯(lián)通，這里的“各種數(shù)據(jù)”是兩個維度上的：

各種存儲介質(zhì)，比如OSS

各種數(shù)據(jù)格式， 比如文本文件，視頻，圖像，音頻，基因，氣象等格式的數(shù)據(jù)

而數(shù)據(jù)的這兩個維度的特征，都是通過EXTERNAL TABLE的概念來引入MaxCompute的計算體系的。 與讀取OSS數(shù)據(jù)的使用方法類似，對OSS數(shù)據(jù)進行寫操作，在如上打開安全授權(quán)通道后，也是先通過CREATE EXTERNAL TABLE語句創(chuàng)建出一個外部表，再通過標準MaxCompute SQL的INSERT INTO/OVERWRITE等語句來實現(xiàn)的，這里先用MaxCompute內(nèi)置的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，并將前面提到的兩個維度的外部數(shù)據(jù)信息關(guān)聯(lián)到這個外部表上。

數(shù)據(jù)存儲介質(zhì)： LOCATION 將一個OSS上的地址關(guān)聯(lián)到外部表上，對這個外部表的進行讀寫操作都會反映到這個OSS地址上。

數(shù)據(jù)存儲格式： StorageHandler用來表明對這些數(shù)據(jù)的讀寫操作方式，這里使用了MaxCompute內(nèi)置的?com.aliyun.odps.TsvStorageHandler， 用戶可以使用這個由系統(tǒng)自帶的實現(xiàn)來讀取和寫出TSV文件。 同時用戶也可以通過MaxCompute的SDK來自定義StorageHandler, 這個將在后面的章節(jié)介紹。

其中OSS數(shù)據(jù)存儲的具體地址的URI格式為：

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

最后還要提到的是，在上面的DDL語句中定義了外部表的Schema, 對于數(shù)據(jù)輸出而言，這表示輸出的數(shù)據(jù)格式將由這個Schema描述。 就TSV格式而言，這個schema描述比較直觀容易理解； 而在用戶自定義的輸出數(shù)據(jù)格式上，這個schema與輸出數(shù)據(jù)的聯(lián)系則更松散一些，有著更大的自由度。 在后面介紹通過自定義StorageHandler/Outputer的時候會詳細展開。

2.2 通過對External Table的 INSERT 操作實現(xiàn)TSV文本文件的寫出

在將OSS數(shù)據(jù)通過External Table關(guān)聯(lián)上后，對OSS文件的寫出可以對External Table做標準的SQL INSERT OVERWRITE/INSERT INTO來操作。 具體輸出數(shù)據(jù)的來源可以有兩種

數(shù)據(jù)源為MaxCompute的內(nèi)部表： 也就是說可以通過對外表INSERT操作來實現(xiàn)MaxCompute內(nèi)部表數(shù)據(jù)到外部存儲介質(zhì)的寫出。

數(shù)據(jù)源為之前通過External Table引入MaxCompute計算體系的外部數(shù)據(jù)： 這可以用來將外部數(shù)據(jù)引入MaxCompute進行計算，然后再存儲到（不同的）外部存儲地址，或者甚至是不同的外部存儲介質(zhì)（比如將TableStore數(shù)據(jù)經(jīng)由MaxCompute導出到OSS）。

2.2.1 從MaxCompute內(nèi)部表輸出數(shù)據(jù)到OSS

這里先來看第一種場景：假設(shè)我們已經(jīng)有一個名為tpch_lineitem的MaxCompute內(nèi)部表，其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。 現(xiàn)在我們希望將其中的一部分數(shù)據(jù)以TSV格式導出到OSS上面。 那么在用上述DDL創(chuàng)建出External Table之后，使用如下INSERT OVERWRITE操作就可以實現(xiàn)：

INSERT OVERWRITE TABLE tpch_lineitem_tsv_external SELECT l_orderkey, l_suppkey, l_discount, l_tax, l_shipdate, l_linestatus, l_shipmode, l_commentFROM tpch_lineitemWHERE l_discount = 0.07 and l_tax = 0.01;

這里將從內(nèi)部的tpch_lineitem表中，在符合l_discount = 0.07 并 l_tax = 0.01的行中選出8個列（對應(yīng)tpch_lineitem_tsv_external這個外部表的schema）按照TSV的格式寫到OSS上。 在上面這個INSERT OVERWRITE操作成功完成后，就可以看到OSS上的對應(yīng)LOCATION產(chǎn)生了一系列文件：

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文件夾下產(chǎn)生了一個.odps文件夾，這其中將有一些.tsv文件，以及一個.meta文件。 這樣子的文件結(jié)構(gòu)是MaxCompute(ODPS)往OSS上輸出所特有的：

通過MaxCompute對一個OSS地址，使用INSERT INTO/OVERWRITE 外部表來做寫出操作，所有的數(shù)據(jù)將在指定的LOCATION下的.odps文件夾產(chǎn)生；

其中.odps文件夾中的.meta文件為MaxCompute額外寫出的宏數(shù)據(jù)文件，其中用于記錄當前文件夾中有效的數(shù)據(jù)。 正常情況下，如果INSERT操作成功完成的話，可以認為當前文件夾的所有數(shù)據(jù)均是有效數(shù)據(jù)。 只有在有作業(yè)失敗的情況下需要對這個宏數(shù)據(jù)進行解析。?即使是在作業(yè)中途失敗或被kill的情況下，對于INSERT OVERWRITE操作，再跑一次成功即可。 如果對于高級用戶，一定需要解析.meta文件的話，可以聯(lián)系MaxCompute技術(shù)團隊。

這里迅速看一下這些tsv文件的內(nèi)容：

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上產(chǎn)生了對應(yīng)的TSV數(shù)據(jù)文件。

最后，大家可能也注意到了，這個INSERT OVERWRITE操作產(chǎn)生了多個TSV文件，對于MaxCompute內(nèi)置的TSV/CSV處理來說，產(chǎn)生的文件數(shù)目與對應(yīng)SQL stage的并發(fā)度是相同的，在上面這個例子中，INSER OVERWITE ... SELECT ... FROM ...; 的操作在源數(shù)據(jù)表(tpch_lineitem) 上分配了1000個mapper，所以最后產(chǎn)生了1000個TSV文件的。 如果需要控制TSV文件的數(shù)目，可以配合MaxCompute的各種靈活語義和配置來實現(xiàn)。 比如如果需要強制產(chǎn)生一個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后，現(xiàn)在同樣的數(shù)據(jù)只了一個輸出TSV文件，當然這個文件的size就大多了。 這方面的調(diào)控技巧還有很多，都是可以依賴SQL語言的靈活性，數(shù)據(jù)本身的特性，以及MaxCompute計算相關(guān)設(shè)置來實現(xiàn)的，這里就不深入展開了。

2.2.2 以MaxCompute為計算介質(zhì)，實現(xiàn)不同存儲介質(zhì)之間的數(shù)據(jù)轉(zhuǎn)移

External Table作為一個MaxCompute與外部存儲介質(zhì)的一個切入點，之前已經(jīng)介紹過對OSS數(shù)據(jù)的讀取以及TableStore數(shù)據(jù)的操作，結(jié)合對外部數(shù)據(jù)讀取和寫出的功能，就可以實現(xiàn)通過External Table實現(xiàn)各種各樣的數(shù)據(jù)計算/存儲鏈路，比如：

讀取External Table A關(guān)聯(lián)的OSS數(shù)據(jù)，在MaxCompute上做復雜計算處理，并輸出到External Table B關(guān)聯(lián)的OSS地址

讀取External Table A關(guān)聯(lián)的TableStore數(shù)據(jù)，在MaxCompute上做復雜計算處理，并輸出到External Table B關(guān)聯(lián)的OSS地址

而這些操作與上面數(shù)據(jù)源為MaxCompute內(nèi)部表的場景，?唯一的區(qū)別只是SELECT的來源變成一個External table，而不是MaxCompute內(nèi)置表。

3. 通過自定義StorageHandler來實現(xiàn)數(shù)據(jù)輸出

除了使用內(nèi)置的StorageHandler來實現(xiàn)在OSS上輸出TSV/CSV等常見文本格式，MaxCompute非結(jié)構(gòu)化框架提供了通用的SDK，允許用戶對外輸出自定義數(shù)據(jù)格式文件，包括圖像，音頻，視頻等等。 這種對于用戶自定義的完全非結(jié)構(gòu)化數(shù)據(jù)格式支持，也是MaxCompute從結(jié)構(gòu)化/文本類數(shù)據(jù)的一個向外擴展，在這里我們會以一個圖像處理的例子，來走通整個【OSS->MaxCompute->OSS】數(shù)據(jù)鏈路，尤其著重介紹對OSS輸出文件的功能。

為了方便大家理解，這里先提供一個在使用用戶自定義代碼的場景下，數(shù)據(jù)在MaxCompute計算平臺上的流程：

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

外部的（OSS）數(shù)據(jù)經(jīng)過非結(jié)構(gòu)化框架轉(zhuǎn)換，會使用Java用戶容易理解的InputStream類提供給自定義代碼接口。 用戶自實現(xiàn)Extract邏輯只需要負責對輸入的InputStream做讀取/解析/轉(zhuǎn)化/計算，最終返回MaxCompute計算平臺通用的Record格式；

這些Record可以自由的參與MaxCompute的SQL邏輯運算，這一部分計算是基于MaxCompute內(nèi)置的強大結(jié)構(gòu)化SQL運算引擎，并可能產(chǎn)生新的Record

運算過后的Record中再傳遞給用戶自定義的Output邏輯，用戶在這里可以進行進一步的計算轉(zhuǎn)換，并最終將Record里面需要輸出的信息通過系統(tǒng)提供的OutputStream輸出，由系統(tǒng)負責寫到OSS。

值得指出的是，這里面所有的步驟都是可以由用戶根據(jù)需要來進行自由的選擇與拼接的。 比如如果用戶的輸入就是MaxCompute的內(nèi)部表，那步驟1.就沒有必要了，事實上在前面的章節(jié)2中的例子，我們就實現(xiàn)了將內(nèi)部表直接寫成OSS上的TSV文件的流程。 同理， 如果用戶沒有輸出的需求，步驟3. 就沒有必要，比如我們之前介紹的OSS數(shù)據(jù)的讀取。 最后，步驟2.也是可以省略的，比如如果用戶的所有計算邏輯都是在自定義的Extract/Output中完成，沒有進行SQL邏輯運算的需要，那步驟1.是可以直接連接到步驟3.的。

理解了上面這個數(shù)據(jù)變換的流程，我們就可以來通過一個圖像處理例子來看看怎么具體的通過非結(jié)構(gòu)化框架在MaxCompute SQL上完整的實現(xiàn)非結(jié)構(gòu)化數(shù)據(jù)的讀取，計算以及輸出了：

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

這里我們先提供實現(xiàn)這整個【OSS->MaxCompute->OSS】數(shù)據(jù)鏈路需要用到的MaxCompute SQL query，并做簡單的注解，詳細的用戶代碼實現(xiàn)邏輯將在后面的3.2子章節(jié)中介紹SDK接口的時候做展開解釋。

3.1.1 關(guān)聯(lián)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)

說明：

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

通過SERDEPROPERTIES來實現(xiàn)參數(shù)傳遞，格式為'key'='value', 具體用法可以參見基本功能介紹以及下面的用戶代碼說明

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

指定包含用戶JAR包，內(nèi)含自定義的StorageHandler/Extractor/Outputer，以及需要的三方庫（這里用到了Java ImageIO庫，具體見下面用戶代碼范例）。JAR包通過ADD JAR命令上傳，可以參見基本功能介紹。

另外要說明的是這里指定的External Table的schema就是用戶在進行Extract操作后構(gòu)造的Record格式，具體怎么構(gòu)造這個Schema用戶可以根據(jù)需要自己根據(jù)能從輸入數(shù)據(jù)中抽取到的信息定義。 在這里我們定義了對于輸入圖片數(shù)據(jù)，會將圖片名稱，圖片的長和寬，以及圖片的二進制bytes抽取出來放進Record（見后面的Extractor代碼說明），所以就有了上面的【STRING，BIGINT，BIGINT，BINARY】的schema。

3.1.2 關(guān)聯(lián)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';

說明： 可以看到這里創(chuàng)建關(guān)聯(lián)輸出圖像文件的External Table，使用的DDL語句，與前面關(guān)聯(lián)輸入圖像時使用的DDL語句是非常類似的：只是LOCATION不一樣，表明圖像數(shù)據(jù)處理后將輸出到另外一個地址。 另外還有一點就是這里我們沒有使用SERDEPROPERTIES來進行傳參，這個只是在這個場景上沒有需求，在有需求的時候可以用同樣的方法把參數(shù)傳遞給outputer。 當然這里兩個DDL語句如此相似，有一個原因是因為我們這個例子中用戶代碼中對于Extract出的Record以及輸入給Outputer的Record使用了一樣的schema, 同時這一對Extractor和Outputer都被封裝在了同一個ImageStorageHandler里放在同一個JAR包里。?在實際應(yīng)用中，這些都是可以根據(jù)實際需求自己調(diào)整的，由用戶自己選擇組合和打包方式。

3.1.3 從OSS讀取原始圖片數(shù)據(jù)到MaxCompute, 計算處理，并輸出圖像到OSS

在上面的3.1.1以及3.1.2子章節(jié)中的兩個DDL語句，分別實現(xiàn)了把輸入OSS數(shù)據(jù)，以及計劃輸出OSS數(shù)據(jù)，分別綁定到兩個LOCATION以及指定對應(yīng)的用戶處理代碼，參數(shù)等設(shè)置。 然而這兩個DDL語句對系統(tǒng)而言，只是進行了一些宏數(shù)據(jù)的記錄操作，并不會涉及具體的數(shù)據(jù)計算操作。 在這兩個DDL語句運行成功后，運行如下SQL語句才會引發(fā)真正的運算。 換句話說，在Fig.1中描述的整個【OSS->MaxCompute->OSS】數(shù)據(jù)讀取/計算/輸出鏈路，實際上都是通過下面一個簡單的SQL 語句完成的：

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

這看起來就是一個標準的MaxCompute SQL語句，只不過因為涉及了images_output_external和images_input_external這兩個外部表，所以真正進行的物理操作與傳統(tǒng)的SQL操作會有一些區(qū)別：在這個過程中，涉及了讀寫OSS，以及通過ImageStorageHandler這個wrapper，調(diào)用自定義的Extractor，Outputer代碼來對數(shù)據(jù)進行操作。 下面就來具體看看在這個例子中的用戶自定義代碼實現(xiàn)了怎樣的功能，以及具體是如何實現(xiàn)的。

3.2 ImageStorageHandler實現(xiàn)

如同之前介紹過的，MaxCompute非結(jié)構(gòu)化框架通過StorageHandler這個接口來描述對各種數(shù)據(jù)存儲格式的處理。 具體來說，StorageHandler作為一個wrapper class, 讓用戶指定自己定義的Exatractor(用于數(shù)據(jù)的讀入，解析，處理等) 以及Outputer(用于數(shù)據(jù)的處理和輸出等)。 用戶自定義的StorageHandler 應(yīng)該繼承?OdpsStorageHandler，實現(xiàn)getExtractorClass以及getOutputerClass?兩個接口。

通常作為wrapper class, StorageHandler的實現(xiàn)都很簡單，比如這里的ImageStorageHandler?就只是通過這兩個接口指定了我們將使用ImageExtractor以及ImageOutputer:

package com.aliyun.odps.udf.example.image;public class ImageStorageHandler extends OdpsStorageHandler {@Overridepublic Class<? extends Extractor> getExtractorClass() {return ImageExtractor.class;}@Overridepublic Class<? extends Outputer> getOutputerClass() {return ImageOutputer.class;} }

另外要說明的是如果確定在使用某個StorageHandler的時候，只需要用到Extractor，或者只需要用到Outputer功能，那不需要的接口則不用實現(xiàn)。 比如如果我們只需要讀取OSS數(shù)據(jù)而不需要做INSERT操作，那getOutputerClass()的實現(xiàn)只需要扔個NotImplemented exception就可以了，不會被調(diào)用到。

3.3 ImageExtractor實現(xiàn)

因為對于SDK中Extractor接口的介紹以及對用戶如何寫一個自定義的Extractor，在之前介紹的OSS數(shù)據(jù)的讀取中已經(jīng)有所涉及，所以這里就不再對這方面做深入的介紹。

Extractor的工作在于讀取輸入數(shù)據(jù)并進行用戶自定義處理，那么我們首先來看看這里由images_input_external這個外表綁定的OSS輸入LOCATION上存放的具體數(shù)據(jù)內(nèi)容：

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圖像數(shù)據(jù)，分辨率從 400 x 400 到 1200 x 1200不等。 具體在這個例子中用到的ImageExtractor的詳細代碼在github上可以找到, 這里只做一些簡單介紹說明該Extractor做了些什么工作：

從輸入的OSS地址上使用非結(jié)構(gòu)化框架提供的InputStream接口讀取圖像數(shù)據(jù)，并在本地進行如下操作

• 對于圖像寬度小于1024的圖片，統(tǒng)一放大到1024 x 1024; 對于圖像寬度大于1024的圖片，跳過不進行處理
• 處理過的圖片，在內(nèi)存中轉(zhuǎn)存成由輸入?yún)?shù)指定的格式（JPG)

把處理后在內(nèi)存中的JPG數(shù)據(jù)的原始字節(jié)存入輸出的Record中的BINARY field, 同一個Record中還將存放處理后圖像的長和寬（都是1024）, 以及原始的圖像名字（這個可以從輸入的InputStream上獲取）；

填充后的Record從Extract接口返回進入MaxCompute系統(tǒng)；

在這個過程中，用戶可以靈活的進行各種操作，比如額外的參數(shù)驗證等。

另外要說明的是，目前Record作為MaxCompute結(jié)構(gòu)化數(shù)據(jù)處理的基本單元，有一些額外的限制，比如BINARY/STRING類型都有8MB大小的限制，但是在大部分場景下這個大小應(yīng)該是能滿足存儲需求的。

3.4 ImageOutputer的實現(xiàn)

接下來我們著重講一下ImageOutputer的實現(xiàn)。 首先所有的用戶輸出邏輯都必須實現(xiàn)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中只會調(diào)用一次。 用戶可以在setup里面做初始化準備工作，另外通常需要把setup()傳遞進來的這三個參數(shù)保存成ouputerd的class variable, 方便之后output()或者close()接口中使用。 而close()這個接口用于方便用戶代碼的掃尾工作。

通常情況下大部分的數(shù)據(jù)處理發(fā)生在output(Record)這個接口內(nèi)。 MaxCompute系統(tǒng)會根據(jù)當前outputer分配處理的Record數(shù)目不斷調(diào)用，也就是對每個輸入Record系統(tǒng)會調(diào)用一次?output(Record)。 系統(tǒng)假設(shè)在一個output(Record) 調(diào)用返回的時候，用戶代碼已經(jīng)消費完這個Record, 因此在當前output(Record)返回后，系統(tǒng)可能將這個Record所使用的內(nèi)存用作它用： 所以不推薦一個Record中的信息在跨多個output()函數(shù)調(diào)用被使用，如果一定有這個需求的話，用戶必須把相關(guān)信息通過class variable等方式自行另外保存。

3.4.1 ImageOutputer.setup()

setup用于初始化整個outputer, 在這個接口上提供了整個outputer操作過程中可能需要的參數(shù)：

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

在我們這個ImageOutputer里，setup()的實現(xiàn)比較簡單：

@Overridepublic 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的實現(xiàn)，可以按照OutputStream使用)，兩個接口唯一的區(qū)別是next()獲取的OutputStream寫出的文件名完全由MaxCompute系統(tǒng)決定，而next(String fileNamePostfix)則允許用戶提供文件名的postfix。 提供這個postfix的意義是，在輸出文件具體地址和名字格式總體由MaxCompute系統(tǒng)決定的前提下，用戶依然可以定制一個方便理解的postfix。 比如使用next("_boat.jpg")?得到的OutputStream可能對應(yīng)如下一個輸出文文件：

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

這其中尾端的"_boat.jpg"可以幫助用戶理解輸出文件的涵義。 如果這個?OutputStream是由next()獲得的話，那對應(yīng)的輸出文件可能就是這樣的：

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

用戶可能就需要具體讀取這個文件才能知道這個文件中具體存放了什么內(nèi)容。

前面提到output(Record)這個接口會由系統(tǒng)不斷調(diào)用，但是應(yīng)該強調(diào)的是，并不一定在每一個Record都需要調(diào)用一次OutputStreamSet.next()接口來獲得一個新的OutputStream。?事實上在大多數(shù)情況下，我們建議在一個Outputer里面盡可能減少調(diào)用next()的次數(shù)（最好只調(diào)用一次）。 也就是說理想情況下，一個outpuer只應(yīng)該產(chǎn)生一個輸出文件。 比如處理TSV這種文本格式文件，假設(shè)有5000個Record對應(yīng)5000行TSV數(shù)據(jù)，那么最理想的情況是應(yīng)該把這5000行數(shù)據(jù)全部寫到一個TSV文件中。 當然用戶可能會有各種各樣不同的切分輸出文件的需求：比如希望每個文件大小控制在一定范圍，或比如文件的邊界有顯著的意義等等。

具體到當前這個圖像例子，從下面的ImageOutputer代碼實現(xiàn)中可以看出，這個例子中確實是處理每個Record就調(diào)用一次next()的，因為在當前場景中，每一個輸入的Record都表示一張圖片的信息（binary bytes, 圖像名字，圖像長寬），所以這里通過多次調(diào)用next()來輸出多個圖片文件。 但是我們還是需要再次強調(diào)，調(diào)用next()的次數(shù)過多可能有一些其他弊端，比如造成碎片化小數(shù)據(jù)在OSS上的存儲等等。 尤其在MaxCompute這種分布式計算系統(tǒng)上，因為系統(tǒng)本身就會調(diào)度起多個outputer進行并行計算處理，如果每個outpuer都輸出過多文件的話，最后產(chǎn)生的文件數(shù)目會有一個乘性效應(yīng)。 回頭來看我們這個例子中，即使在這里，多個圖像其實也可以通過一個OutputStream，按照tar/tar.gz的方式寫到單個文件中，這些都是在實現(xiàn)具體系統(tǒng)中用戶需要根據(jù)自己的場景, 以及處理邏輯，輸出數(shù)據(jù)類型等信息來進行優(yōu)化和tradeoff的。

在理解了這些之后，現(xiàn)在來具體看看ImageOutputer的實現(xiàn)output接口實現(xiàn)：

@Overridepublic 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);}

可以看到這里主要就做了三件事情：

根據(jù)之前保存的圖像名字，長寬信息，和編碼方式(".jpg")拼出一個帶擴展名的輸出文件名postfix。

讀取圖像binary bytes，并用getEdgeImage()來利用sobel算子對圖像做邊緣檢測。 具體getEdgeImage()的實現(xiàn)這里就不進行深入解釋了： 使用了標準的sobel模板卷積算法， 有興趣看ImageOutputer源碼即可。

對每一個圖像產(chǎn)生一個新的OutputStream并將數(shù)據(jù)寫出，至此當前Record處理完畢，寫出一張圖片到OSS，output()函數(shù)返回。

3.4.3 ImageOutputer.close()

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

至此我們就介紹完了一個output的實現(xiàn)，現(xiàn)在可以看看在運行完這個SQL query，對應(yīng)OSS地址的數(shù)據(jù)：

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 ...

可以看到圖像數(shù)據(jù)按照期待格式寫到了指定地址，這里我們就選一個輸入圖像(lena.bmp)以及對應(yīng)的輸出圖像(M1_0_-1--1-7-lena_1024x1024.jpg)看一下對比：

這個例子中整個圖像處理流程已經(jīng)通過如上的SQL query完成。 而從上面展示的ImageExtractor以及ImageOutputer?源代碼，我們可以看出整個過程中用戶的邏輯基本與寫單機圖像處理程序無異，用戶的代碼只需要在Extractor上做InputStream到Record的準換，而在Outputer上做反向的Record到OutputSteam的寫出處理，其他核心的處理邏輯實現(xiàn)基本和單機算法實現(xiàn)相同，在用戶的層面，并不用去操心底層分布式系統(tǒng)的細節(jié)以及MaxCompute和OSS的交互。

3.5 數(shù)據(jù)處理步驟的靈活性

從上面這個例子中我們也可以看出，在一個完整的【OSS->MaxCompute->OSS】數(shù)據(jù)流程中，Extractor和Outputer中涉及的具體計算邏輯其實也并不一定會有一個非常明確的邊界。 Extractor和Outputer只要各自完成所需的轉(zhuǎn)換Record/Stream的轉(zhuǎn)換，具體的額外算法邏輯在兩個地方都有機會完成。 比如上面這個例子的整個流程涉及了如下圖像處理相關(guān)的運算：

圖像的縮放 （統(tǒng)一到 1024 x 1024）

圖像格式的轉(zhuǎn)換 (BMP -> JPG)

圖像的Sobel邊緣檢測

上面的例子實現(xiàn)中，把1. 和 2. 放在ImageExtractor中完成，而3.則放在ImageOutputer中完成，但并不是唯一的選擇。 我們完全可以把所有3個步驟都放在ImageExtractor中完成，讓ImageOutputer只做Record到寫出最后圖像的操作；也可以在ImageExtractor中只做讀取原始binary到Recrod, 而把所有3個圖像處理步驟都放在ImageOutputer中進行，等等。 具體進行怎樣的選擇，用戶可以完全根據(jù)需要自己實現(xiàn)。

另外一個系統(tǒng)設(shè)計的點是如果對于一個數(shù)據(jù)需要做重復的運算，那可以考慮將數(shù)據(jù)從OSS中通過Extractor讀出進MaxCompute，然后存儲成MaxCompute的內(nèi)置表格再進行（多次）的計算。 這個對于MaxCompute和OSS沒有進行混布，不在一個物理網(wǎng)絡(luò)上的場景尤其有意義： MaxCompute從內(nèi)置表中讀取數(shù)據(jù)無疑要比從外部OSS存儲服務(wù)中讀出數(shù)據(jù)要有效得多。 在上面3.1.3子章節(jié)中的圖像處理例子，這個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;

通過把數(shù)據(jù)寫到一個內(nèi)部images_internal表中，后面如果有多次讀取數(shù)據(jù)的需求的話，就可以不再去訪問外部OSS了。 這里也可以看到MaxCompute非結(jié)構(gòu)化框架以及SQL語法本身提供了非常高的靈活性和可擴展性，用戶可以根據(jù)實際計算的不同模式/場景/需求，來在上面完成各種各樣的數(shù)據(jù)計算工作流。

5. 結(jié)語

非結(jié)構(gòu)化數(shù)據(jù)處理框架隨著MaxCompute 2.0一起推出，意在豐富MaxCompute平臺的數(shù)據(jù)處理生態(tài)，來打通阿里云核心計算平臺與阿里云各個重要存儲服務(wù)之間的數(shù)據(jù)鏈路。 在之前介紹過的讀取OSS以及處理TableStore數(shù)據(jù)的整體方案后，本文側(cè)重介紹數(shù)據(jù)往OSS的輸出方案，并依托一個圖像處理的處理實例，展示了【OSS->MaxCompute->OSS】整個數(shù)據(jù)鏈路的實現(xiàn)。 在這些新功能的基礎(chǔ)上，我們希望實現(xiàn)整個阿里云計算與數(shù)據(jù)的生態(tài)融合： 在不同的項目上，我們已經(jīng)看到了在MaxCompute上處理OSS上的海量視頻，圖像等非結(jié)構(gòu)化數(shù)據(jù)的巨大潛力。 今后隨著這個生態(tài)的豐富，我們期望OSS數(shù)據(jù)，TableStore數(shù)據(jù)以及MaxCompute內(nèi)部存儲的數(shù)據(jù)，都能在MaxCompute的核心計算引擎上進行融合，從而產(chǎn)生更大的價值。

原文鏈接

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

總結(jié)

以上是生活随笔為你收集整理的MaxCompute与OSS非结构化数据读写互通（及图像处理实例）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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