统计脚本一.十亿级找出N天前数据

作者: john不哭 | 来源:发表于2019-10-04 21:40 被阅读0次

统计脚本一.十亿级找出N天前数据
全长转录组-subread下机数据统计N50
20220218和你一起成长日记125
Elasticsearch Top hits Aggregati
liunx非gui环境做统计图
数据分析师P1系列课程-张宇晖
2019-06-03
Hive 通过 UDAF 实现 `分组取前 Top N`
脚本练习1
SHELL删除N天前数据

前言

写统计脚本系列，是把工作中的一些需求记录下来，并且写下优化手段。算是一种技术总结。

需求描述
从一个支持亚马逊s3协议的资源池中，下载该池特定的业务的元信息。然后找出所有30天符合业务逻辑的条目，用于执行删除。
下载代码略过。大概操作如下：池中文件前缀为 abc/hash-code 其中abc为业务线，对应了资源池生命周期(为什么不用桶生命周期管理，の，池的管理并发慢于文件的增加速度，需要手动管理。(｡•́︿•̀｡))。
文件列表下载难点在于：

数据库中无法获取历史列表，数据库严格按照周期删除。
只能通过资源池访问日志或者是资源池元数据中获取文件列表。然而池的元数据aws sdk中只有单线程，死循环下载的方法(每次获取一千条，需要下载12亿条，测试下载3亿条该方法需要40几小时)。利用历史访问日志获取所有文件id，历史文件下载，保存需要4T空间，浪费大量资源，并且大量文件id已经被删除过了，放弃日志方案。

文件下载优化手段(改良aws s3 getbucklist方法)：

第一层拆分，每一个桶对应不同的线程。
第二层拆分，通过s3 sdk中setPrefix 方法，业务拆分不同生命周期，不同hash-code范围的方式，进一步提高桶内并发。拆分例子如下
abc10/00000019-493b-473a-b90c-d4423c3fe3bd
abc 为业务线，后面的数字为保存日期，hash域拆分为00... 01...。这样拆分后并发达到了桶 * 业务线 * 保存日期 * 256。
12亿条，200G数据，100个线程的线程池，3小时即可拉取所有数据。

找出30天前的数据(利用字符串代替时间转换，将查找时间从1900秒优化到400秒，从cpu瓶颈到io瓶颈，python为例)。
基准环境如下：
查找数据量12亿条，200G。输出结果，3.8亿条，50g。测试机器单盘2t sas，54核，256g ram，进程池大小为48，内存使用大概为10G。
未优化前做法：源数据为多个文件(大小大致相等)，实现无锁，多进程(python需要多进程)消费，字符串中时间转为内部数据结构后对比。观察单进程cpu使用达到100%，io wait 忽略不计，遂优化cpu使用。代码如下（ps必须先把文件夹建立好，直接执行，会出现冲突。多个进程同时创建文件，会导致一个进程失败。）：

# -*- coding: utf-8 -*-
import datetime as dt
import os
from concurrent.futures import ThreadPoolExecutor
from  multiprocessing import Pool
import time


input_size = 256*1024*1024
def get_path():
    path_list = []
    with open("path_list", "r") as f:
        temp = f.readlines()
        for each in temp:
            size, path = each.split("\t")
            if size !="0":
                path_list.append(path.strip("\n"))
    return path_list

def makedir(file):
    for each in file
        try:
            start_time = dt.datetime(year=2019,month=8,day=11,hour=0,minute=0,second=0)
            output_path = file.replace("family","family/temp")
            dir_path = "/".join(output_path.split("/")[:-1])
            if os.path.exists(dir_path) == False:
                os.makedirs(dir_path)
        except Exception as e:
            print str(e)

def process(file):
      try:
        with open(file, "r") as f, open(output_path+"output", "w") as output:
            pass
            judge = f.readlines(input_size)
            output_buff = []
            day_ago = 10
            if "H10" in file:
                day_ago = 10
            elif "H20" in file:
                day_ago = 20
            elif "H30" in file:
                day_ago = 30
            elif "H40" in file:
                day_ago = 40
            elif "H50" in file:
                day_ago = 50
            elif "H60" in file:
                day_ago = 60
            while judge != None and judge != []:
                for each in judge:
                    key, etag, bucket, modify_date, size  = each.split("\t")
                    if dt.datetime.strptime(modify_date, "%a %b %d %H:%M:%S CST %Y")<start_time-dt.timedelta(days=day_ago):
                        output_buff.append(each)
                output.writelines(output_buff)
                output_buff = []
                judge = f.readlines(input_size)

    except Exception as e:
        print str(e)


def bulk_execute(method, method_par):
    start = time.time()
    pool = Pool(48)
    pool.map(method, method_par)
    pool.close()
    pool.join() 
    print time.time()-start
   
    
    


def run():
    # process()
    path_list = get_path()
    makedir(path_list)
    bulk_execute(process, path_list)

if __name__ == "__main__":
    run()

优化：生成一个保留日期字典，直接对比字符串。观察单核进程cpu使用下降一半，部分进程出现io wait ，磁盘io接近上限((ÒωÓױ),突然有种冲动，想将磁盘退回，拆分为多盘，提高io。•̥́ ˍ •̀)。代码如下：

# -*- coding: utf-8 -*-
import datetime as dt
import os
from concurrent.futures import ThreadPoolExecutor
from  multiprocessing import Pool
import time


input_size = 256*1024*1024
def get_path():
    path_list = []
    with open("path_list", "r") as f:
        temp = f.readlines()
        for each in temp:
            size, path = each.split("\t")
            if size !="0":
                path_list.append(path.strip("\n"))
    return path_list

def makedir(file):
    for each in file
        try:
            start_time = dt.datetime(year=2019,month=8,day=11,hour=0,minute=0,second=0)
            output_path = file.replace("family","family/temp")
            dir_path = "/".join(output_path.split("/")[:-1])
            if os.path.exists(dir_path) == False:
                os.makedirs(dir_path)
        except Exception as e:
            print str(e)
    
def process(file):
    try:
        with open(file, "r") as f, open(output_path+"output", "w") as output:
            pass
            judge = f.readlines(input_size)
            output_buff = []
            day_ago = 10
            if "H10" in file:
                day_ago = 10
            elif "H20" in file:
                day_ago = 20
            elif "H30" in file:
                day_ago = 30
            elif "H40" in file:
                day_ago = 40
            elif "H50" in file:
                day_ago = 50
            elif "H60" in file:
                day_ago = 60
            
            # 将时间对比变为字符串对比
            ban_dict = set([(dt.datetime(2019,9,14)-dt.timedelta(each)).strftime("%a %b %d %H:%M:%S CST %Y")[:10] for each in range(day_ago+1+34) ])
            while judge != None and judge != []:
                for each in judge:
                    key, etag, bucket, modify_date, size  = each.split("\t")
                    if modify_date[:10] not in ban_dict:
                        output_buff.append(each)
                output.writelines(output_buff)
                output_buff = []
                judge = f.readlines(input_size)

    except Exception as e:
        print str(e)



def bulk_execute(method, method_par):
    start = time.time()
    pool = Pool(48)
    pool.map(method, method_par)
    pool.close()
    pool.join() 
    print time.time()-start
   
    
    


def run():
    # process()
    path_list = get_path()
    makedir(path_list)
    bulk_execute(process, path_list)

if __name__ == "__main__":
    run()