首頁

物聯(lián)資訊

低代碼數(shù)據(jù)平臺BI大屏定制相關文章

大數(shù)據(jù)中的數(shù)據(jù)傾斜及解決辦法

作者：速優(yōu)云數(shù)據(jù)平臺架構師

閱讀數(shù)：290

更新時間：2024-02-23 20:34:50

id="%E4%B8%80%E4%BB%80%E4%B9%88%E6%98%AF%E6%95%B0%E6%8D%AE%E5%80%BE%E6%96%9C" class="code-line" dir="auto" data-line="0">一、什么是數(shù)據(jù)傾斜

數(shù)據(jù)傾斜指mapreduce計算架構或flink等流式計算平臺下，在進行計算時，發(fā)生的數(shù)據(jù)操作卡在某一個子計算任務而導致整個任務被卡住的現(xiàn)象。例如寫hive SQL計算指標時，發(fā)現(xiàn)數(shù)據(jù)從1%到99%很快，確一直卡在99%很長一段時間而無法完成任務。

實際上，發(fā)生這樣的原因是，在計算一條SQL的時候，某個任務被分配的key太多了。從而導致，其他任務很快完成任務，但是某個任務確一直在計算，造成『一人累死，其他人閑死』的情況。

二、MapReduce原理

什么是map任務？什么是recude任務呢？

我們知道hive的底層是通過HDFS將文件數(shù)據(jù)存在磁盤上，數(shù)據(jù)以key、value的形式存儲，而map操作相當于將把key、value鍵值對讀取出來，重新組合。然后將整理好的數(shù)據(jù)，交給recude，reduce進行聚合計算。

舉個例子，如下數(shù)據(jù)：（HDFS文件存儲數(shù)據(jù)時，存在行存儲和列存儲兩種，下面以列存儲為例）

HDFS中存儲文件的格式如下：

（0,"hive spark hive hbase" ) （21,"hadoop hive spark hive" ) （39,"sqoop flume scala scala" )

map操作將對HDFS文件進行分隔，并將每一行分成一個的<key,value>值（初始的k值是根據(jù)偏移量來的），然后在將<key,value>值，轉成如下的格式，并存入緩存中：

(hive,1),(spark,1),(hive,1),(hbase,1);
(hadoop,1),(hive,1),(spark,1), (hive,1);
(sqoop,1),(flume,1),(scala,1), (scala,1);

map操作將不同分區(qū)中的key值進行整合排序，存放到一個集合當中，格式如下：

(hive,[1,1,1,1]),(spark,[1,1]),(hbase,[1]),(hadoop,[1]),(sqoop,[1]),(flume,[1]),(scala,[1,1])

隨后，Reduce任務先處理多個Map任務的輸出結果，再根據(jù)分區(qū)將其分配到不同的Reduce節(jié)點上（這個過程就是shuffle）；Reduce任務對多個Map的輸出結果進行合并、排序、計算，生成新的 (k,v) 值，具體如下：

(hive,4),(spark,2), (scala,2) ,(hbase,1),(hadoop,1),(sqoop,1),(flume,1)

Reduce任務會將上一步輸出的<k,v>寫到HDFS中，生成文件。

三、數(shù)據(jù)傾斜解決方案

數(shù)據(jù)傾斜一般都發(fā)生在reduce階段。

reduce階段最容易發(fā)生傾斜的操作是join和count distinct

下面列舉幾個常見的hive數(shù)據(jù)傾斜場景，和其對應解決方案：

3.1 空值數(shù)據(jù)傾斜優(yōu)化

在上報的日志信息中，通常會出現(xiàn)信息丟失的情況，如果用上報缺失的字段去關聯(lián)相關字段時就會出現(xiàn)數(shù)據(jù)傾斜的問題。

案例：日志中的user_id上報缺失，如果取其中的user_id和用戶表的user_id進行關聯(lián)的時候就會出現(xiàn)數(shù)據(jù)傾斜。

解決辦法：數(shù)據(jù)傾斜主要原因是join的key值發(fā)生傾斜，key值包含很多空值或是異常值，通常的做法是，對空值或者異常值賦一個隨機的值來分散key。

select * from log a
left join user b 
on 
    case when (a.user_id is null or a.user_id = '-' or user_id='0')

    --空值和異常值處理
then 
    concat('sql_hive',rand()) else a.user_id end = b.user_id

3.2 大表和小表的優(yōu)化

在hive SQL的join操作中，mr過程是按照join的key進行分發(fā)，而在join左邊的表的數(shù)據(jù)會首先被讀進內存，如果左邊表的key相對分散，讀入內存的數(shù)據(jù)會比較小，join任務就會執(zhí)行的比較快；如果左邊的key比較集中并且數(shù)據(jù)量比較大時，數(shù)據(jù)傾斜就會比較嚴重，執(zhí)行時間將會增加。

經驗總結：為了能夠避免數(shù)據(jù)傾斜現(xiàn)象，通常將數(shù)據(jù)量小的表放在join的左邊，此外，還需要使用map join讓小維度的表先進內存，在map完成reduce操作。

SQL操作：/*+MAPJOIN(smallTable)*/

Set hive.auto.convert.join=true
Set hive.mapjoin.smalltable.filesize=25000000
select /*+MAPJOIN(b)*/ 
    a.key,a.value
from a
join b on a.key = b.key
-- a為大表，b為小表

上面操作主要是將小表全部讀入內存中，在map階段大表的每個map和小表進行匹配，節(jié)省了reduce階段的時間，提高了數(shù)據(jù)執(zhí)行效率。

3.3 大表和大表的join優(yōu)化

情況一：Map階段輸出的key數(shù)量上，導致reduce階段的reduce數(shù)量為1

案例：日志表中的user_id和用戶表關聯(lián)，user表上有500W+條記錄，把user分發(fā)到所有的map開銷很大，并且map join不支持大表操作。用普通的join操作，數(shù)據(jù)傾斜會產生。

解決辦法：這類問題產生的根本原因就是數(shù)據(jù)業(yè)務特性強，兩個表都是大表。因此我們可以針對性的削減業(yè)務過程，比如log表中user_id有上百萬個，但是每天會員的uv不會太多，有交易的會員不會很多，有點擊的會員數(shù)不會很多等，在業(yè)務過程削減冗余的數(shù)據(jù)量，避免數(shù)據(jù)傾斜的發(fā)生。
-- 思路：先過濾出需要被使用的唯一的user_id，再計算相關指標 select /*MAPJOIN(b)*/ from log a left join ( select /*MAPJOIN(b)*/ b1.* from (select distinct user_id from log) b join user b1 on b.user_id = b1.user_id ) a1 on a.user_id = a1.user_id;

情況二：Map輸出key分布不均勻，商品信息表的key對應大量的value，導致數(shù)據(jù)傾斜。

案例：商品信息表a中的信息填充到商品瀏覽日志表b中，使用商品id進行關聯(lián)。但是某些人買商品瀏覽量較大，造成數(shù)據(jù)偏移。

解決辦法：熱點數(shù)據(jù)和非熱點數(shù)據(jù)拆分處理
select * from ( select /*MAPJOIN(i)*/ a.id, a.time, a.amount, b.name, b.loc, b.cat from a left join (select * from a where uid in ('1001','1002')) as b -- 熱點子表 on a.uid = b.uid where a.uid in('1001','1002') -- 熱點數(shù)據(jù) ) union all select a.id, a.time, a.amount, b.name, b.loc, b.cat from a left join b on a.uid = b.uid where a.uid is not in('1001','1002') -- 非熱點數(shù)據(jù)

3.4 count(distinct)數(shù)據(jù)傾斜優(yōu)化

場景：在多個維度的同一個度量的count distinct

案例：根據(jù)月份和性別，統(tǒng)計買家的1月份男顧客數(shù)，女顧客數(shù)

原始方案：

SELECT 
 seller,
 COUNT(DISTINCT CASE WHEN month=1 AND sex = 'M' THEN buyer END) M01_BUYER_CNT,
 COUNT(DISTINCT CASE WHEN month=1 AND sex = 'F' THEN buyer END) M01_FEMALE_BUYER_CNT
FROM 
 SHOP_ORDER
 where 
  log_date = '20220301'
 group by 
 seller

改造方案：

把DISTINCT用到的buyer，也加到group by統(tǒng)計上，然后再進行業(yè)務計算

with t1 as( -- 第一步：group by統(tǒng)計，結果存到t1中
    select 
        buyer,
        seller,
        count(case when month=1 and sex = 'M' then buyer end) s_M01_male_buyer_cnt,
        count(case when month=1 and sex = 'F' then buyer end) s_M01_female_buyer_cnt
    from 
       SHOP_ORDER
    where 
        log_date = '20220301'
    group by 
    seller,
    buyer
  )
  select  -- 聚合目標指標
      seller,
      sum(case when s_M01_male_buyer_cnt>0 then 1 else 0 end) as s_M01_male_buyer_cnt,
      sum(case when s_M01_female_buyer_cnt>0 then 1 else 0 end) as s_M01_female_buyer_cnt
  from 
    t1
  group by 
    seller