hdfs 文件统计

hdfs、hive用一段时间之后，我们会想要知道文件系统里的文件哪些是经常被使用的，哪些是长时间没有被使用。

如果是用mapreduce或者spark生成的文件，会有文件打小不均，并且远大于或者远小于一个块打小的情况。

namenode记录了每个文件的创建时间和最后读取时间，文件的路径和文件尺寸。为了拉取hdfs上的文件的创建时间、读取时间、文件尺寸，创建了如下python脚本.

脚本使用snakebite来调用namenode上的api（使用namenode的thrift协议）。
代码含义可以看代码中的注释。代码把文件的信息写入csv并上传到hdfs，在hive中创建表，把分析的事交给hive来处理。

#!/usr/bin/env python

import os
import sys
import time

from snakebite.client import AutoConfigClient as HDFSClient


data_dir = sys.argv[1] # hdfs目录
table_dir = sys.argv[2] # 表的目录
table_name = sys.argv[3] # 表的名称
today = time.strftime('%Y-%m-%d') # 今天的日期

# 生成snakebite客户端
client = HDFSClient()

def stat_hdfs_dir(dirname):
    """ 统计 dirname """
    data_files = []
    data_prefix = 'd' + dirname.rstrip('/').replace('/', '_')
    data_file_prefix = os.path.join(data_dir, data_prefix)
    fp = None
    lines = 0


    for f in client.ls([dirname], recurse=True): # 递归便利目录下的文件
        if fp is None:
            data_file_path = '%s.%02d.csv' % (data_file_prefix, len(data_files))
            data_files.append(data_file_path)
            fp = open(data_file_path, 'w') # 打开一个文件用于写入
            lines = 0


        if f['file_type'] == 'f': # 只统计文件
            at = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(int(f["access_time"]/1000))) # 最后访问时间
            mt = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(int(f["modification_time"]/1000))) # 创建时间
            fp.write('%s,%d,%s,%s\n' % (f['path'], f['length'], at, mt))
            lines += 1

            if lines >= 1000000: # 因文件数很多，所以每100万行换一个新文件
                fp.close()
                lines = 0
                fp = None

    if fp is not None:
        fp.close()

    # 在hdfs上准备一个目录，用于上传结果文件，每天一个目录
    hdfs_data_dir = os.path.join(table_dir, 'day=%s' % today, 'statdir=%s' % data_prefix)
    if client.test(hdfs_data_dir, exists=True):
        print('hdfs data dir %s exists' % hdfs_data_dir)
        print(list(client.delete([hdfs_data_dir], recurse=True)))
    else:
        print('hdfs data dir %s not exists' % hdfs_data_dir)
    print(list(client.mkdir([hdfs_data_dir], create_parent=True)))


    # snakebite can not upload file， 上传文件，调用hdfs命令上传
    os.system('hdfs dfs -put %s %s' % (' '.join(data_files), hdfs_data_dir))
    # 把刚上传完的文件加入到表
    os.system('hive -e "alter table %s add IF NOT EXISTS partition (day=\'%s\', statdir=\'%s\') location \'%s\'"' % (table_name, today, data_prefix, hdfs_data_dir))


stat_dirs = ['/data’] # 需要分析的目录


if __name__ == '__main__':
    for d in stat_dirs:
        stat_hdfs_dir(d)

hive表结构如下

我统计了目录下文件的数量、总打小、小文件（小于20M）数量，大文件（大于200M）数量，从未使用的文件数量，15天未使用文件数量，30天未使用文件数量. cluster_manage.file_stat是hive上的表名。
sql如下

SELECT a.file_cnt,
       a.file_size,
       a.small_cnt,
       a.large_cnt,
       a.not_used,
       a.not_used_15d,
       a.not_used_31d,
       a.`day`,
       a.statdir,
       a.file_size/a.file_cnt/1024/1024 AS avg_size,
       a.small_cnt / a.file_cnt AS small_percent,
       a.large_cnt / a.file_cnt AS large_percent,
       a.not_used / a.file_cnt AS not_used_percent,
       a.not_used_15d / a.file_cnt AS not_used_15d_percent,
       a.not_used_31d / a.file_cnt AS not_used_31d_percent
FROM
  (SELECT count(1) AS file_cnt,
          sum(`size`)/1024/1024 AS file_size,
          sum(if(`size` < 20*1024*1024, 1, 0)) AS small_cnt,
          sum(if(`size` > 200*1024*1024, 1, 0)) AS large_cnt,
          sum(if(access_time <= modification_time, 1, 0)) AS not_used,
          sum(if(datediff(current_timestamp(), access_time) BETWEEN 15 AND 30, 1, 0)) AS not_used_15d,
          sum(if(datediff(current_timestamp(), access_time) > 30, 1, 0)) AS not_used_31d,
          `day`,
          statdir
   FROM cluster_manage.file_stat
   WHERE `day`='2020-09-24'
     AND path NOT RLIKE '/_SUCCESS$'
   GROUP BY `day`,
            statdir) a ;