1、前言

在離線數據研發中，隨著業務的快速發展以及業務復雜度的不斷提高，數據量的不斷增長，尤其得物這種業務的高速增長，必然帶來數據邏輯復雜度的提升，數據量越大，復雜度越高，對任務的性能的要求就越高，因此，任務性能的優化就成了大家必然的話題，在離線數倉招聘中，這幾乎成了必考題目。

大數據領域，為了提高超大數據量的計算性能，幾代人不斷在努力，不斷榨取著計算機的CPU、內存、磁盤每一個模塊的性能，從早期的縱向擴展（提升計算機性能，如IBM、ORACLE 早期推崇的服務器到小型機到大型機的演進）到目前的大規模橫向擴展(分布式集群模式），都是旨在提升大數據的性能。

本文重點從在分布式計算模式下，如何來優化任務，大家耳熟能詳的常見優化如：mapjoin skewjoin distribute by 等就不多做贅述，本文主要探索技巧、策略及方法。

2、任務優化策略

2.1 優化方向

補充說明：目前得物大數據在阿里云的dataworks 環境下，集群層面做了比較多的工作，IO、網絡、機架感應等暫時無需過多關注，如有自建集群時，可重點關注，我們重點關注JOIN  和REDUCE 層面，優化細節也重點基于這兩個方向做細節展開。

2.2 優化手段

對于優化手段優化方法，我們大多數習慣性從技術手段出發，更多的從算子、邏輯兼容等來處理，但是在某些業務場景下，如埋點日志,數據量一般比較大，這種情況無論技術手段如何干預，都無法解決存儲和計算帶來的資源消耗，這時候如果要提升SLA，就得從業務場景出發，做好業務的分類分級以及核心數據分流，因此，本文的優化手段會從技術手段和業務手段兩方面展開。

• 技術手段

聚焦于技術手段來處理任務，參加上述單點任務優化方向，主要是SQL 邏輯、模型規范、算子優化及可能存在的集群優化

• 業務手段

聚焦于業務特性、業務邏輯來進行處理，基于不同的業務特性及重要程度，從生產、采集、模型、數據消費全鏈路進行梳理和架構優化，同時形成一套數據鏈路上的通知及約束機制，避免上游變更帶來的下游數據故障及恢復問題。

3、優化實踐案例

優化策略中，定義好優化方向、優化手段，接下來，我們選取一些比較有效的沉淀出來的方案，展開講講如何來做任務優化。

前文講述，目前的得物的數據平臺特性（dataworks)，我們在IO、網絡、RPC 通信機制等暫時涉入不深，且對于面向業務的數據研發來言，大部分人不會過多關注底層的實現原理，暫不做過多深入探討。

我們基于上面方向中的技術手段講述幾個日常常見的優化案例

3.1 數據重分發（Distribute &Rand)

3.1.1 數據重分發的要點

日常數據研發中，最常見的且使用較多的就是數據傾斜或數據量帶來的數據重分發（打散或隨機），對于數據的重分發，主要分以下幾點：

• 優化小文件
• 數據傾斜
• 排序&隨機

小文件過多帶來的MAP 端資源損耗和數據傾斜是我們日常開發過程中最為常見的性能問題，而這兩點大多跟rand()隨機數有一定的關系，通過數據分發和打散和規避掉大部分此場景下的問題。

數據重分發一般代碼操作如下所示

select c1,c2... from tablename distribute by c1[,...]
select c1,c2... from tablename distribute by rand([,seed])[,...]

對于rand() 我們要注意幾點，可讓我們在優化任務時，知其然，更知其所以然。

• rand() 隨機數的生成規律跟數學概率有莫大的關系，尤其在算法中，會被經常性問到，給定隨機生成的N個數，構造等概率事件的發生器，跑題了，繼續說回在hive 或odps 場景下，rand() 函數是隨機生成的0-1 的double 類型的數字。
• rand(int seed) 函數可以根據種子參數，構造一個穩定的隨機值，加上種子參數，得到的結果是相對穩定的，尤其在處理小文件過程中，這一步很重要。
• Hive 和odps 場景中，隨機函數多與pmod()、mod()、floor()、ceil() 等函數結合使用，可以根據不同的業務場景，來構造任意范圍內的隨機整數，比如在處理數據重分發解決數據傾斜的問題時，同時擔心影響這種重分發帶來過多的小文件，隨機數可以這樣來取  floor(rand())*N/ceil(rand())+1,取1-N 之間的整數。

比如在流量數據里面，因為大量空值時，結合rand函數，解決數據傾斜問題：

select * 
from  a 
left join b on a.order_id = nvl(b.order_id ,concat('hive',rand()))
--b中的order_id 存在大量空值 的時候

3.1.2 數據重分發的作用

對于數據重分發，我們主要是用來對處理數據結果進行小文件合并以及對數據處理中的傾斜問題進行優化。在大多數的處理中，我們習慣于使用Distribute by Rand() *N 的方式，其實這個方式可能存在問題，在處理類似問題時候，我們可以選擇基于seed種子的Rand函數，來維持隨機數的穩定性。這里需要知曉，distribute by 實際上做了一次shuffle的分發，默認是按照給定key進行的hash操作（可以理解為一次repartion重新分區），這里面是可以進行定制分區邏輯的，可以通過重寫hive當中partition的接口，實現不同策略的重分發。

• 處理小文件合并

使用方式一：指定固定分發列，做一次shuffle的merge操作，DEMO如下：

SELECT column1, column2,column.... FROM TABLEX WHERE ds = '${bizdate}'DISTRIBUTE BY '${bizdate}',columns1....

使用方式二：指定給定的文件數,這里要用到rand()函數了，一般有兩種寫法：

第一種寫法（上文討論過，這種寫法在一定情況下會出現數據問題）：

SELECT column1, column2,column.... FROM TABLEX WHERE ds = '${bizdate}'DISTRIBUTE BY FLOOR(RAND()*N)/CEIL(RAND()*N)

第二種寫法（加隨機種子，產生穩定的隨機序列）：

SELECT column1,column2,column.... FROM (    SELECT column1, column2,column...., FLOOR(RAND(seed)*N) AS rep_partion FROM      TABLEX      WHERE ds = '${bizdate}')DISTRIBUTE BY rep_partion

• 處理JOIN中的傾斜：與上述邏輯同理，主要是借助一次分發，使得需要shuffle的數據能在一個節點進行數據處理。

3.2 數據膨脹(Explode)

在join過程中，我們之前提到了一種基于BLOOMFILTER算法的優化方法。在某些情況下，當join的表中出現一個表的量級很大，另外一個表無法mapjoin切熱鍵key在概率分布上呈現隨機性，這個時候就可以在一定程度上，對較小表中的join key進行一定程度的膨脹，由于join的發生是在reduce階段，因此可以構造出穩定的多條主鍵，在不同的reduce中對數據進行jion操作，進而一定程度上解決join傾斜帶來的問題。基本原理如下圖所示：

一個小例子，當研發使用數組形式存儲數據(sku_ids)時，數倉想要拿到數組中每一個sku_id，使用 lateral view EXPLODE。代碼如下：

select order_id 
from a 
lateral view explode(split(order_ids,',')) v1 as order_id 
group by order_id

結果展示：

order_ids  order_id 
101,102,103 101
101,102,103 102
101,102,103 103
104,105 104
104,105 105

目前，膨脹函數已經有開發出來有現成的UDTF函數來支持，可以支撐任意膨脹量級的數據進行膨脹。只需要構造膨脹區間對應的隨機函數即可，還是需要用到Rand()函數來實現。

數據膨脹方式帶來的問題：

在解決了數據傾斜重新打散的問題之后，在計算層面會增加一定的數據計算量。此外，如果能基于分桶進行二次索引分片，也可以在引擎側考慮基于該方向的自適應傾斜優化。

3.3 數據分桶(Bucket)

在數據量比較大的情況下，單表數據做分區會存在下游使用效率上的限制，而數據在某些列上（或者構造業務列）存在高度聚集，或者存在可以優化提升的巨大空間，在此時，我們就可以對列進行散列分桶，在分區的基礎上進行桶表的設計，桶上可以對應索引向量，將極大的提升數據使用上的效率。

在數據隨機抽樣、JOIN場景中，也會極大的提升整個數據的計算性能和效率。在hive中，該功能默認是關閉的，需要set hive.enforce.bucketing=true打開支持，odps 下可能無需特別關注，需要注意一般而言，桶的個數將與一次作業中對應的reduce數量一致。

其實，基于分桶的邏輯，在引擎側可以做更多的優化（比如引擎側可以優化分桶存儲的策略）。在join中，根據索引進行join層面的動態優化，在超大數據join過程中，基于桶進行單位數據的本地優化等等都是可以做非常多的優化操作的，由于在目前的業務場景中，較少用到數據分桶，因此這里不做更深入的拓展，詳細的可以自行百度，查看關于桶表的使用，更進一步，合理分桶，加上排序后的索引，能高效優化單表查詢使用的效率。

3.4 并發與并行控制

在計算機入門的時候，我們就經常聽到并發與并行，線程與進程等概念。而在數據研發中，我們發現，其實對于整個作業來說，同樣遵循類似的調優規則。一般的，一個作業最大的map數是9999，reduce數最大是1000。雖然可以提高單個任務吞吐量，但是會消耗更長的時間和資源調度上的等待。另一方面，當完成一個同類作業，往往需要多個任務進行，如果任務下面可以多個作業并行處理，單個作業也能夠并發執行，那么就能夠更大程度地榨取整個集群的資源，從而達到突破計算瓶頸和上線的目的。目前在開源HADOOP體系中，我們沒有腳本模式來支持靈活的任務自動分配和調度，但是可以采用SHELL/PYTHON腳本+SQL的方式來實現這一目的，其實借助猛犸調度在一定范圍內也能達到同樣的效果。

3.5 多路輸出與物化(Read Once Output More)

這個部分我們主要談談HIVE（spark）的CTE寫法（WITH...AS...）以及From語法的應用。這兩個語法，在日常開發稍微復雜的任務時候，可以大大清晰整個復雜SQL的邏輯，同時，在多路讀寫中，通過物化的方式還能在一定程度上加速作業的運行。

• CTE(with.... as ...)使用

• 基本使用非常簡單，cte的語法主要是為了提高代碼的可讀性，雖然在整個性能的優化上未必達到很好的效果，但是在一定程度上，能大大提高任務的邏輯清晰度。很多時候，我們在多個邏輯過程中，通過臨時表的方式進行任務的串行，使用with...as...能達到類似的效果。同時with...as...可以深層嵌套，因此是比較好的一種選擇方式。無論是線上任務還是視圖，都可以使用CTE的寫法——目前比較遺憾的是HIVE的CTE目前不支持遞歸。

代碼示例(可以使用多個with,抽出代碼片段)：

with a as (
    select * from test1
    where xxx = xxx
)
,
b as (
    select * from a
)
select * from b limit 100;

• 物化設置

由于with...as...等同于一個SQL片段，下文中會多次引用該片段的別名，相當于視圖的味道。所以，這里面使用是一個虛擬的概念，實際上只是邏輯生效，實際運行是則是翻譯成實際的MR邏輯去執行，如果下游引用該SQL片段較多，這時候MR執行會多次掃描原始數據，執行多次相同的MR操作邏輯,此時，就可以在第一次執行中來物化CTE寫法中定義的SQL片段,從而達到優化的目的。在hive之前的版本中，該功能是默認關閉的，可以通過下面參數來開啟,在新的hive版本中，該功能是默認開啟,但是默認引用次數是3次。

社區版hive 如下所示，我們的ODPS 下，大家無需太多關注，這部分做技術擴展和了解即可。

• FROM使用（一讀多寫）

• FROM也是本人在實際研發中遇到多路輸出時采用比較多的一種手段之一。當有多個不同的分區，或者多個不同的目標輸出，或者有多個不同的子邏輯的過程中，可以將主邏輯全部開發完成，然后再進行多路輸出。多路輸出操作的使用限制如下：
• 單條 multi insert語句中最多可以寫255路輸出。超過255路，會上報語法錯誤。
• 單條 multi insert語句中，對于分區表，同一個目標分區不允許出現多次。
• 單條 multi insert語句中，對于非分區表，該表不能出現多次。

比如在流量業務場景時，需要寫動態分區，就可以使用from，一個代碼小例子：

from (
   select aa,bb,pt,sec_pt from test
)
insert OVERWRITE table du_temp.temp_01 partition (pt = 'xx'，sec_pt = 'test1' )
select aa,bb where sec_pt = 'test1'
insert OVERWRITE table du_temp.temp_01 partition (pt = 'xx'，sec_pt = 'test2' )
select aa,bb where sec_pt = 'test2'

4、思考&總結

在數據研發領域，數據的技術手段無論多么豐富，平臺發展何等完善，都不能說能解決業務的所有問題。一定是先有業務，才會有對應的問題。在面對大數據量，高時效性，高復雜計算的場景，我們需要結合業務的特性，模型的改造，鏈路的設計，甚至打破常規等方式來產出不同的方案。在另一個方面，數據研發的工作也遠遠不是單點問題的解決和兜底，相反需要各方的配合與共同的智慧。