建模杂谈系列215 ADBS Update V3

说明

本来在完成这次量化实验之前不想再改版，但想想至少还有3个ADBS要在本次中完成，算下来改所花的时间还是少于因为不改而做的额外配置，所以还是改。

本次升级要解决几个问题：

• 1 Redis Var的配置化管理
• 2 Worker 的命令行配置(分发与非分发）

内容

1 Redis Var

目前看来，Redis Var有这么几种角色/功能：

• 1 通用依赖。例如像时间轴这种的变量，每个worker都要使用。
• 2 项目定制变量。每个项目都有自己流程的缓存变量，目前有一些项目的缓存还是不正确的，但是只是用于监控，也就放过了。

所以本次的改变是将所有的redis变量都集中在config文件里，有一些只和项目相关，使用上又是固定的变量，就直接用项目名称去格式化就好。

所以会涉及到所有的app和monitor,逐个扫过一遍，确保通用的redis_var会被集中管理。

2 Worker

Worker的分发模式(fetch)有助于各Worker进行并行执行，而顺序取(range)则可以保证每一条数据不会漏掉。

这两种模式区别非常明显，又都会用到，所以把这个参数通过执行命令的参数透到外面的命令行会比较灵活。

具体的执行Worker是逻辑性问题，不需要再修改。

3 操作

• 1 先创建一个新的文件夹，用于存放新的模板文件
• 2 以docker run -it 方式打开老的镜像

3.1 configs_base.py

先把最新的配置内容粘贴过来，然后逐个查看每个app中需要使用的redis变量。

# 通用Redis变量 - 默认都是本地redis
global_redis_var = {}
global_redis_var['redis_agent_host'] = redis_agent_host
global_redis_var['redis_connection_hash'] = redis_connection_hash
global_redis_var['some_var'] = None

# 项目用的appRedis变量 - 默认都是本地redis
app_redis_var = {}
app_redis_var['redis_agent_host'] = redis_agent_host
app_redis_var['redis_connection_hash'] = redis_connection_hash
app_redis_var['app01_PullToStep1MongoIn'] = 'BUFF.%s.step1_mongo_in.pf.app.app01_PullToStep1MongoIn.af.gp.0.uf.flow_count' % project_name
app_redis_var['app03_PullToStep1MongoOut'] = 'BUFF.%s.step1_mongo_in.pf.app.app01_PullToStep1MongoIn.af.gp.0.uf.flow_count' % project_name

# 监控用的Redis变量
monitor_redis_var = {}
monitor_redis_var['redis_agent_host'] = redis_agent_host
monitor_redis_var['redis_connection_hash'] = redis_connection_hash
monitor_redis_var['monitor03_Step1TasksClaimed'] ='BUFF.%s.step1_mongo_in.pf.app.monitor03_Step1TasksClaimed.af.gp.0.uf.last_stat' % project_name
monitor_redis_var['monitor04_StreamIn_count'] = 'BUFF.%s.step1_mongo_in.pf.app.app01_PullToStep1MongoIn.af.gp.0.uf.flow_count' % project_name

# 数值型转换（如果声明的变量缺少流程就会中断）
app01_toDoubleVarList = []
app03_toDoubleVarList = []

然后是app需要使用的redis变量，这些应该都是项目相关的固定变量。目前看来，app01是需要redis变量的。

3.2 app01_PullToStep1MongoIn.py

先修改这个。需要使用redis var来记录入系统流量，同时也需要维持数值化变量的任务。如果可以全部是数值的话就不用管。

..
from configs_base import global_redis_var,app_redis_var

..
toDoubleVarList = app01_config['toDoubleVarList']

..

import time 
redis_var =app_redis_var['app01_PullToStep1MongoIn']
redis_buff = req.post(redis_agent_host + 'getv/',json ={'k':redis_var}).json()['data']

3.3 app03_PullToStep1MongoOut.py

app03并不需要提供其他的统计，无需redis变量，但是要增加对于数值型变量的转换功能。

..
from configs_base import global_redis_var,app_redis_var

..
toDoubleVarList = app03_config['toDoubleVarList']

3.4 monitor01_StreamFlow.py

这个monitor主要是看项目默认的输入输出队列，不必改。

3.5 monitor02_DB_Recs.py

这个是看项目相关数据库的总条数，也不用改。

3.6 monitor03_Step1TasksClaimed.py

这个是监控入系统数据被分配的情况。为了减少Mongo的统计压力，采用了增量统计的方式，使用了一个redis变量记录上次的统计点。

这个也属于“项目间差异命名，项目内固定命名”的变量，所以也提出来，放在configs_base里定义，在这个程序体中直接引入就可以。

...

from configs_base import global_redis_var,monitor_redis_var

...

redis_var = monitor_redis_var['monitor03_Step1TasksClaimed']

3.7 monitor04_StreamIn_count.py

这是用于统计入系统流量的，同样需要修改redis变量。monitor04在上个版本已经改过，现在把redis_var的导入集中管理。

3.8 worker01_Go.py

之前会给这个worker01进行不同的命名，发现在多数情况下要定制的部分其实不多，所以可以做一个约定，将这个步骤的调度都命名为 worker01_Go.py,而对应的Worker，不管实现什么功能，都将对应的输出命名为： worker_af, worker_Chain_session_list 以及worker_Chain_session_dict。Worker的名称叫TheWorker.py

启动时，以argparse的方式对其进行调度，默认参数可以维持现状(这样就不必动sche.py)。

argparse的示例如下：

import argparse
# =================  解析参数
def get_arg():
    parser = argparse.ArgumentParser(description='Customized Arguments')
#     parser.add_argument('-p','--pkl', default='Meta')

    parser.add_argument('--is_fetch_mode')

    # 准备解析参数
    args = parser.parse_args()

    is_fetch_mode = args.is_fetch_mode

    return is_fetch_mode

if __name__ =='__main__':
    is_fetch_mode = get_arg() or False
    if is_fetch_mode is not False:
        is_fetch_mode = True 
    print('is_fetch_mode :', is_fetch_mode)

调用：

root@3b2281835e89:/workspace# python3 test.py
is_fetch_mode : False
root@3b2281835e89:/workspace# python3 test.py --is_fetch_mode=xx
is_fetch_mode : True

重写worker.py

import os 

print('>>>work is Running  ')
runcode =''
for some_app in ['worker01_Go.py']:
    runcode += str(os.system('python3 %s' % some_app))
print('>>>work RunCode :%s ' % runcode)

新建worker01_Go.py

...
import argparse
# =================  解析参数
def get_arg():
    parser = argparse.ArgumentParser(description='Customized Arguments')
#     parser.add_argument('-p','--pkl', default='Meta')

    parser.add_argument('--is_fetch_mode')
    parser.add_argument('--worker_name')
    parser.add_argument('--group_name')

    # 准备解析参数
    args = parser.parse_args()

    is_fetch_mode = args.is_fetch_mode
    worker_name = args.worker_name
    group_name = args.group_name

    return is_fetch_mode,worker_name,group_name


is_fetch_mode,worker_name,group_name = get_arg()

# 参数映射1
if is_fetch_mode is not None:
    is_fetch_mode = True 
else:
    is_fetch_mode = False
# 参数映射2
worker_name =worker_name or 'alice'

# 参数映射3
# 在init_projects.py定义
group_name =group_name or 'group1'

...
from TheWorker import worker_af, worker_Chain_session_list, worker_Chain_session_dict

这样worker01就可以固定下来不变了,以后每个项目中嵌入自己的worker就可以，worker中对应的实例、链和字典赋给要导入的变量。

如果worker没有定义会怎么样？

整个流是不会有影响的，使用worker是以os方式调用worker01的，如果worker01失败，主程序并不会出错。

3.9 CNT_worker.py

这类调度主要是在冷启动的时候加速处理，启动一定期限的不间断循环，工作结束时对应的容器也会自动销毁。

import os 

import argparse
# =================  解析参数
def get_arg():
    parser = argparse.ArgumentParser(description='Customized Arguments')
#     parser.add_argument('-p','--pkl', default='Meta')

    parser.add_argument('--cnt_limit')
    parser.add_argument('--is_fetch_mode')
    parser.add_argument('--worker_name')
    parser.add_argument('--group_name')

    # 准备解析参数
    args = parser.parse_args()

    cnt_limit = args.cnt_limit
    is_fetch_mode = args.is_fetch_mode
    worker_name = args.worker_name
    group_name = args.group_name

    return cnt_limit,is_fetch_mode,worker_name,group_name


cnt_limit,is_fetch_mode,worker_name,group_name = get_arg()

cnt_limit = cnt_limit or 10000
cnt_limit = int(cnt_limit)

is_fetch_mode = is_fetch_mode or 'yes'
if is_fetch_mode == 'no':
    is_fetch_mode = ''

worker_name = worker_name or ''
group_name = group_name or ''


aleary_run = 0

print('>>>work is Running  ')
for _ in range(cnt_limit):
    aleary_run+=1
    print('Already Run',aleary_run)
    runcode =''
    for some_app in ['worker01_Go.py --is_fetch_mode=%s --worker_name=%s --group_name=%s ' % (is_fetch_mode,worker_name,group_name) ]:
        runcode += str(os.system('python3 %s' % some_app))
    print('>>>work RunCode :%s ' % runcode)

3.10 Band_worker.py

这类调度主要是为了增加的实时数据的处理，在时间内一直循环，时间到时对应的容器也会自动销毁。Band_worker因为会常态运行，所以加一个参数，让其稍微停顿一下，一般一秒一次。

import os 

import time
def get_time_str1(ts = None,bias_hours=0):
    ts = ts or time.time()
    return time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(ts + bias_hours*3600))

import argparse
# =================  解析参数
def get_arg():
    parser = argparse.ArgumentParser(description='Customized Arguments')
#     parser.add_argument('-p','--pkl', default='Meta')

    parser.add_argument('--to_dt')
    parser.add_argument('--pace')
    parser.add_argument('--is_fetch_mode')
    parser.add_argument('--worker_name')
    parser.add_argument('--group_name')

    # 准备解析参数
    args = parser.parse_args()

    to_dt = args.to_dt
    pace = args.pace
    is_fetch_mode = args.is_fetch_mode
    worker_name = args.worker_name
    group_name = args.group_name

    return to_dt,pace,is_fetch_mode,worker_name,group_name


to_dt,pace,is_fetch_mode,worker_name,group_name = get_arg()

to_dt = to_dt or '2099-01-01 00:00:00'

pace = pace or 10000
pace = int(pace)

is_fetch_mode = is_fetch_mode or 'yes'
if is_fetch_mode == 'no':
    is_fetch_mode = ''

worker_name = worker_name or ''
group_name = group_name or ''



aleary_run = 0

print('>>>work is Running  ')
while True:
    aleary_run+=1
    print('Already Run',aleary_run)
    runcode =''
    for some_app in ['worker01_Go.py --is_fetch_mode=%s --worker_name=%s --group_name=%s ' % (is_fetch_mode,worker_name,group_name) ]:
        runcode += str(os.system('python3 %s' % some_app))
    print('>>>work RunCode :%s ' % runcode)

    if get_time_str1() >= to_dt:
        break
    if pace > 0:
        time.sleep(pace)

4 提交更改，生成新镜像

略