仓库

Hadoop 平台「瘦身计划」- 数据清理

2022-05-17  本文已影响0人  六层楼那么高

Hadoop 平台运行至今，前期处于放任自由的状态，后期才开始稍加管控，指定相关数据使用规范。日积月累，数据规模越来越大，元数据暴增，Namenode rpc 频繁超时告警，Namenode HA 切换也较为频繁。

企业的预算不可能无限增加，所以一个良好的平台规范以及定时数据清理机制，对平台的来说至关重要，属于非常实在的降本增效工作。

这里记录下我们组的近期制定的 HDFS「瘦身计划」。

1. HDFS「瘦身计划」

  1. 通过脚本代码，分析导出以下四类数据，按目录导出成文件
  2. 联系各个业务，依据我们提供的数据，进行清理工作
    • 目录不存在和空表：项目里面有空表和表对应的 hdfs 路径不存在的情况，需要删除
    • N个月零增长：项目里面有 N 个月没有数据变化的表，需要按需删除
    • 未压缩的表：项目里面有数据存在未压缩的情况，需要压缩节省资源
    • 小文件：项目里面包含了大量小文件，需要做小文件合并**

2. 操作手册

  1. 目录不存在和空表：确认后，直接删除 hive 表或表 hdfs 目录
  2. 三个月零增长：删除表，如需保留需备注原因
  3. 未压缩：主要针对非 orc 格式的表，进行：压缩 + 小文件合并，参考章节：2.2 和 2.3
  4. 小文件：对于非 orc 格式的表， 进行：压缩 + 小文件合并, 参考章节：2.2 和 2.3
    对于 orc 格式的表 进行: 小文件合并，参考章节：2.1

备注：小文件合并可以采用非动态分区和动态分区两种方式进行 overwrite 数据。当数据量过大（1亿条以上）时，建议采用非动态分区方式。

2.1 ORC 表小文件处理

（一）orc 非动态分区方式
1、直接覆盖原分区数据，参考：

insert overwrite
  table mgwh_bigvideoprovincessink_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
    mgwh_bigvideoprovincessink_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};"
上一篇下一篇

猜你喜欢

热点阅读