Hadoop 平台运行至今，前期处于放任自由的状态，后期才开始稍加管控，指定相关数据使用规范。日积月累，数据规模越来越大，元数据暴增，Namenode rpc 频繁超时告警，Namenode HA 切换也较为频繁。
企业的预算不可能无限增加，所以一个良好的平台规范以及定时数据清理机制，对平台的来说至关重要，属于非常实在的降本增效工作。
这里记录下我们组的近期制定的 HDFS「瘦身计划」。
1. HDFS「瘦身计划」
- 通过脚本代码，分析导出以下四类数据，按目录导出成文件
- 联系各个业务，依据我们提供的数据，进行清理工作
- 目录不存在和空表：项目里面有空表和表对应的 hdfs 路径不存在的情况，需要删除
- N个月零增长：项目里面有 N 个月没有数据变化的表，需要按需删除
- 未压缩的表：项目里面有数据存在未压缩的情况，需要压缩节省资源
- 小文件：项目里面包含了大量小文件，需要做小文件合并**
2. 操作手册
- 目录不存在和空表：确认后，直接删除 hive 表或表 hdfs 目录
- 三个月零增长：删除表，如需保留需备注原因
- 未压缩：主要针对非 orc 格式的表，进行：压缩 + 小文件合并，参考章节：2.2 和 2.3
-
小文件：对于非 orc 格式的表， 进行：压缩 + 小文件合并, 参考章节：2.2 和 2.3
对于 orc 格式的表 进行: 小文件合并，参考章节：2.1
备注：小文件合并可以采用非动态分区和动态分区两种方式进行 overwrite 数据。当数据量过大（1亿条以上）时，建议采用非动态分区方式。
2.1 ORC 表小文件处理
（一）orc 非动态分区方式
1、直接覆盖原分区数据，参考：
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 动态分区方式
代码参考：
#先设置参数，开启动态分区
#将查询数据插入新的表中
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格式小文件合并，该脚本是在原表基础上操作，用户可以根据实际场景选择创建新表或者基于原表进行小文件合并。
######################################################################################################################
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 非动态分区方式
操作步骤：
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 动态分区方式
操作步骤：
1）压缩（对当前分区进行数据压缩操作）
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）合并（先开启动态分区，将查询结果覆盖当前分区数据）
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();
处理结果：
参考脚本如下
#!/bin/bash
####################################################
#title：小文件合并
#note:orc格式小文件合并,此脚本基于原表操作模式。
####################################################
set -e
#表所在的库名
dbName=`echo "$1" | awk -F '.' '{ print $1 }'`
#表名
tableName=`echo "$1" | awk -F '.' '{ print $2 }'`
#导出表分区信息
hive -e "show partitions ${dbName}.${tableName};" > ${tableName}.txt
#获取分区信息
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`)
#判断表的分区总数
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
{
#获取单个分区（K=V）
partitionValue=`echo ${tablePartitions[i]} | sed 's/\//,and where /'`
#进行数据压缩
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};"
