如何給 HDFS 「減肥」之數(shù)據(jù)清理

Hadoop 平臺運行至今，前期處于放任自由的狀態(tài)，后期才開始稍加管控，指定相關(guān)數(shù)據(jù)使用規(guī)范。日積月累，數(shù)據(jù)規(guī)模越來越大，元數(shù)據(jù)暴增，Namenode rpc 頻繁超時告警，Namenode HA 切換也較為頻繁。

企業(yè)的預算不可能無限增加，所以一個良好的平臺規(guī)范以及定時數(shù)據(jù)清理機制，對平臺的來說至關(guān)重要，屬于非常實在的降本增效工作。

這里記錄下我們組的近期制定的 HDFS「瘦身計劃」。

1. HDFS「瘦身計劃」

  1. 通過腳本代碼，分析導出以下四類數(shù)據(jù)，按目錄導出成文件
  2. 聯(lián)系各個業(yè)務，依據(jù)我們提供的數(shù)據(jù)，進行清理工作
    • 目錄不存在和空表：項目里面有空表和表對應的 hdfs 路徑不存在的情況，需要刪除
    • N個月零增長：項目里面有 N 個月沒有數(shù)據(jù)變化的表，需要按需刪除
    • 未壓縮的表：項目里面有數(shù)據(jù)存在未壓縮的情況，需要壓縮節(jié)省資源
    • 小文件：項目里面包含了大量小文件，需要做小文件合并**

2. 操作手冊

  1. 目錄不存在和空表：確認后，直接刪除 hive 表或表 hdfs 目錄
  2. 三個月零增長：刪除表，如需保留需備注原因
  3. 未壓縮：主要針對非 orc 格式的表，進行：壓縮 + 小文件合并，參考章節(jié)：2.2 和 2.3
  4. 小文件：對于非 orc 格式的表， 進行：壓縮 + 小文件合并, 參考章節(jié)：2.2 和 2.3
    對于 orc 格式的表 進行: 小文件合并，參考章節(jié)：2.1

備注：小文件合并可以采用非動態(tài)分區(qū)和動態(tài)分區(qū)兩種方式進行 overwrite 數(shù)據(jù)。當數(shù)據(jù)量過大（1億條以上）時，建議采用非動態(tài)分區(qū)方式。

2.1 ORC 表小文件處理

（一）orc 非動態(tài)分區(qū)方式
1、直接覆蓋原分區(qū)數(shù)據(jù)，參考：

insert overwrite
  table ods.tmp_owl_newuser_prize_h_ly partition(hourid='2019120101') select
    province_id,
    province_name,
    event_type,
    activity_code,
    activity_name,
    main_site,
    num 
  from
    ods.tmp_owl_newuser_prize_h_ly 
  where
    hourid='2019120101';

（二）orc 動態(tài)分區(qū)方式

代碼參考：

#先設置參數(shù)，開啟動態(tài)分區(qū)
#將查詢數(shù)據(jù)插入新的表中
hive -e "
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.optimize.sort.dynamic.partition=true;
insert overwrite table $tableName partition($partitionName) select * from $tableName where $partitionName='$partitionValue' distribute by rand();"

參考腳本代碼：

#!/bin/bash
######################################################################################################################
#title：小文件合并
#note:orc格式小文件合并，該腳本是在原表基礎上操作，用戶可以根據(jù)實際場景選擇創(chuàng)建新表或者基于原表進行小文件合并。
######################################################################################################################
set -x
set -e
dbName=$1
tableName=$2
partitionName=$3
partitionValue=$4
 
hive -e "
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.optimize.sort.dynamic.partition=true;
use $dbName;
insert overwrite table $tableName partition($partitionName) select * from $tableName where $partitionName='$partitionValue' distribute by rand();
"

2.2. Textfile 表小文件處理

（一）textfile 非動態(tài)分區(qū)方式
操作步驟：
1） 壓縮

set hive.exec.compress.output=false;
set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;
set mapred.output.compression.type=BLOCK;
set hive.merge.size.per.task=256000000;
set hive.merge.smallfiles.avgsize=256000000;
set mapreduce.input.fileinputformat.split.maxsize=256000000;
set mapreduce.input.fileinputformat.split.minsize.per.node=256000000;
set mapreduce.input.fileinputformat.split.minsize.per.rack=256000000;

hive -e "
use $dbName;
insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionName='$partitionValue';"

2） 合并

hive -e "
set hive.exec.compress.output=true;
use $dbName;
insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionName='$partitionValue';"

（二）textfile 動態(tài)分區(qū)方式
操作步驟：
1）壓縮（對當前分區(qū)進行數(shù)據(jù)壓縮操作）

set hive.exec.dynamic.partition.mode=nonstrict;
set hive.optimize.sort.dynamic.partition=true;
 
set hive.exec.compress.output=false;
set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;
set mapred.output.compression.type=BLOCK;
set hive.merge.size.per.task=2560000000;
set hive.merge.smallfiles.avgsize=2560000000; 
set mapreduce.input.fileinputformat.split.maxsize=2560000000;
set mapreduce.input.fileinputformat.split.minsize.per.node=2560000000;
set mapreduce.input.fileinputformat.split.minsize.per.rack=2560000000;
hive -e " 
use $dbName;
insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionName='$partitionValue';"

2）合并（先開啟動態(tài)分區(qū)，將查詢結(jié)果覆蓋當前分區(qū)數(shù)據(jù)）

set hive.exec.dynamic.partition.mode=nonstrict;
set hive.optimize.sort.dynamic.partition=true;
set hive.exec.compress.output=true;
use $dbName;
insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionName='$partitionValue' distribute by rand();

處理結(jié)果：
參考腳本如下

#!/bin/bash
####################################################
#title：小文件合并
#note:orc格式小文件合并,此腳本基于原表操作模式。
####################################################
 
set -e
#表所在的庫名
dbName=`echo "$1" | awk -F '.' '{ print $1 }'`
#表名
tableName=`echo "$1" | awk -F '.' '{ print $2 }'`
 
 
#導出表分區(qū)信息
hive -e "show partitions ${dbName}.${tableName};" > ${tableName}.txt
#獲取分區(qū)信息
partitionName=`cat ${tableName}.txt | sed -n '$p' | awk -F "/" '{for(x=1 ; x<=NF ; x++) print $x}' | awk -F "=" '{print $1F}' | sed '{N;s/\n/,/}'`
 
 
tablePartitions=(`cat ${tableName}.txt`)
#判斷表的分區(qū)總數(shù)
tablePartitionsNum=${#tablePartitions[@]}
 
hive -e "use $dbName;show create table $tableName;" > $tableName.sql
path="/$(grep -A1 LOCATION $tableName.sql | tail -n1 | awk -F "'" '{print $2}' | cut -d"/" -f4-)"
echo "路徑: $path"
 
 
 
for(( i = 0; i < $tablePartitionsNum; i++ )); do
{
        #獲取單個分區(qū)（K=V）
        partitionValue=`echo ${tablePartitions[i]} | sed 's/\//,and where /'`
        #進行數(shù)據(jù)壓縮
 
        echo " insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionValue;"
 
        pathInfo1=`hdfs dfs -count -q -h $path/${tablePartitions[i]}`
        hive -e "
        set hive.exec.dynamic.partition.mode=nonstrict;
        set hive.optimize.sort.dynamic.partition=true;
        set hive.exec.compress.output=false;
 
        set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;
        set mapred.output.compression.type=BLOCK;
        set hive.merge.size.per.task=256000000;
        set hive.merge.smallfiles.avgsize=256000000;
        set mapreduce.input.fileinputformat.split.maxsize=256000000;
        set mapreduce.input.fileinputformat.split.minsize.per.node=256000000;
        set mapreduce.input.fileinputformat.split.minsize.per.rack=256000000;
 
        use $dbName;
 
        insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionValue;
        "
 
        echo " insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionValue distribute by rand();"
        #進行小文件合并
        hive -e "
        set hive.exec.compress.output=true;
        set hive.exec.dynamic.partition.mode=nonstrict;
        set hive.optimize.sort.dynamic.partition=true;
        use $dbName;
 
        insert overwrite table $tableName partition ($partitionName) select * from $tableName where $partitionValue distribute by rand();
        "
 
        echo "壓縮后文件信息"
        hdfs dfs -count -q -v -h $path/${tablePartitions[i]}
        echo "壓縮前文件信息"
        echo $pathInfo1
}
done

2.3 parquet 表小文件處理

壓縮與合并，參考：

hive -e "
set parquet.compression=gzip;
set hive.merge.size.per.task=5120000000; 
set hive.merge.smallfiles.avgsize=5120000000;  
set mapreduce.input.fileinputformat.split.maxsize=5120000000; 
set mapreduce.input.fileinputformat.split.minsize.per.node=5120000000; 
set mapreduce.input.fileinputformat.split.minsize.per.rack=5120000000;
insert overwrite table $tableName partition(dayid=${dayid}) select 
字段
from $tableName where dayid=${dayid};"
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容