在调试的时候转了个圈，发现队列一会有消息，一会没消息。原因是默认的函数里含有了一个另外一个服务器的地址（感觉很像走进科学…)

将sniffer部分进行重写

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

将sniffer和sniffer01_query_akshare在启动时挂载（覆盖）就完成了Sniffer部分。(注意要调整sniffer的执行周期，之前是默认按天的…)

3 适配 app01_PullToStep1MongoIn

主要就是修改一下使用的Redis Buffer的名称,这个是最早设计的时候没有考虑redis buffer 统计入流量。

这里也是同上，修改名字后挂载就可以。

4 透传 TransWorker

在这个Case中，暂时想不到还要做什么ETL，所以可以写一个透传Worker。原来的DummyWorker有些地方写死，并不好，所以做一个微调，这样就可以实现输入的透传了。

明天先实验将这个ADBS运行起来（虽然worker不通，但是可以入库的）

这里碰到一个小问题，就是AETL默认是不允许数据有空值的，这个在当时的设计有当时的考虑(高稳定)，放在这里倒是给自己设了一道坎。

比较合适的解决方式是在sniffer阶段就进行字段的框定和空值的处理，值得注意的是，如果多个数据源都要进入一个ADBS，那么这些数据源只能保持相同的列。

在Sniffer阶段要设计schema，后续的Worker可以透传这个schema。

• Sniffer有一个特殊的变量 vars_to_handle（代码上非必须，流程上必须）。后续Worker会透传其中的内容。

TransWorker.py

from funcs_apifunc_database_model1_6810f9d37e89e5e1f33e1b8f4defa22e import * gs_id = 'rec_id'
sample_listofdict = [{'amt': '3171687.0','close': '4.025','code': '510300','data_dt': '2023-03-09 13:49:00','data_slot': '27972829','data_source': 'AK','high': '4.025','low': '4.024','market': 'SH','open': '4.025','rec_id': 'AK_SH_510300_27972829','vol': '7880','_ch001': 0}]af = APIFunc('apifunc_database_model1', listofdict= sample_listofdict, key_id=gs_id)chain_funcname_dict = {}# 透传
@af.route('/MergeOut', is_force=True)
def MergeOut(input_dict = None, para_dict = None):msg = RuleMSG('MergeOut')# 1 【不检查字段】# need_cols = ['some_str']# input_cols = list(input_dict.keys())# if not (set(need_cols) <= set(input_cols)):#     msg.status = False #     msg.rule_result = None #     msg.msg = 'InputSetError'#     msg.data = None  #     return msg.to_dict()try:msg.msg ='ok'msg.status = Truemsg.data = input_dictmsg.rule_result = 1except:msg.status = Falsemsg.rule_result = Nonemsg.data = Nonereturn msg.to_dict()# >>> 添加到chain_funcname_dict
MergeOut = {}
MergeOut['subline_dict'] =None
MergeOut['main_dict'] = {'key_id':gs_id, 'depend_cols':['data_dt', 'open', 'close', 'high', 'low', 'vol', 'amt', 'data_source', 'code', 'market', 'data_slot'],'input_cols':['data_dt', 'open', 'close', 'high', 'low', 'vol', 'amt', 'data_source', 'code', 'market', 'data_slot']}
MergeOut['para_dict'] = None
chain_funcname_dict['MergeOut'] = MergeOutTransPassChain_session = ['MergeOut']TransPassChain_list = TransPassChain_session
TransPassChain_dict = {}
for k in TransPassChain_list:TransPassChain_dict[k] = chain_funcname_dict[k]if __name__ =='__main__':af.run_chain(chain_funcname_list=TransPassChain_list,chain_funcname_dict=TransPassChain_dict)af.g['ruleresult_frame'][ [x for x in af.g['ruleresult_frame'].columns if x !=gs_id]].sum()

5 修改 app03_PullToStep1MongoOut.py

这里似乎有一个ADBS设计上需要修改的地方：变量类型转换。

原来默认都是字符串，似乎在进行Redis批量写入的时候都转为字符串了。

现在希望能够将变量进行数值类型的转换，因为在使用时，筛选起来比较容易。

在两个地方可以做这个操作，一个是拉到MongoIn的过程，还有一个是拉到MongoOut的过程

保险起见，我们在这两个过程都对数值型的变量进行声明(为Float)

具体的做法就是通过重写一个方法S2M( Stream To Mongo)

...# Pattern: 从队列获取数据，存储到Mongo，然后删除。适用于新增数据。 Stream到Mongo【队列Write相关-删除消息】def S2M(self, stream_name,keyname =None, tier1 = None, tier2 = None,w = None, batch_num = None,sniffer_name = None,log_tier1 = None, log_tier2 = None,time_out = None,redis_agent = None,connection_hash =None,is_return_msg_id_list=False ):must_cols = ['tier1','tier2','sniffer_name','keyname']assert all(must_cols), ','.join(must_cols) + ' 不能为空'cur_w = w or self.w cur_redis_agent = redis_agent or self.redis_agentbatch_num = batch_num or 1msg =''log_tier1 = log_tier1 or tier1 log_tier2 = log_tier2 or 'log_sniffer' + get_time_str2()# 默认三十秒超时time_out = time_out or 30tick1 = time.time()qname = stream_namecur_len_resp = req.post(redis_agent + 'len_of_queue/',json ={'stream_name':qname,'connection_hash':connection_hash}).json()q_len = cur_len_resp['data']print('{} Q has {} Messages' .format (qname,q_len))if q_len :q_data_resp = self.xrange(qname, count = batch_num, redis_agent = redis_agent, connection_hash = connection_hash)q_data = pd.DataFrame(q_data_resp['data'])# q_data = pd.DataFrame(cur_lq.xrange(qname, count=batch_num))msg_id_list = list(q_data['_msg_id'])# 去重q_data1 = q_data.drop_duplicates([keyname], keep='last')# 脱掉自动变量:在这种情况，不需要声明其他参数(通道是默认的_ch001)q_data2 = self.filter_and_add_msg(q_data1)q_listofdict = q_data2.to_dict(orient='records')q_db_resp = cur_w.insert_or_update_with_key(tier1 = tier1, tier2 = tier2, data_listofdict = q_listofdict, key_name=keyname)# 完成后删除消息del_cnt = self.xdel(qname, mid_or_list = msg_id_list)msg = 'ok,deliver(del) {} messages from {}' .format(del_cnt['data'] ,qname)else:msg ='no data source {}' .format(qname)tick2 = time.time()duration = round(tick2 -tick1,2)log_dict = {'sniffer': sniffer_name,'duration':duration,'msg': msg }return cur_w.insert_recs(tier1=log_tier1, tier2=log_tier2, data_listofdict =[log_dict])

要修改的部分其实不多，增加一个toDoubleVarList,将对应的变量做一个转换即可。使用python对象的继承方法，将这个对象

class StreamsIO_X(StreamsIO):# Pattern: 从队列获取数据，存储到Mongo，然后删除。适用于新增数据。 Stream到Mongo【队列Write相关-删除消息】def S2M(self, stream_name,keyname =None, tier1 = None, tier2 = None,w = None, batch_num = None,sniffer_name = None,log_tier1 = None, log_tier2 = None,time_out = None,redis_agent = None,connection_hash =None,is_return_msg_id_list=False, toDoubleVarList = [] ):...# - toDoubleVarListif len(toDoubleVarList):for _var in toDoubleVarList:q_data2[_var] = q_data2[_var].apply(float)

app03_PullToStep1MongoOut.py

...
sio = StreamsIO_X('sio', w = cur_w ,redis_agent = redis_agent_host)
toDoubleVarList = ['open', 'close', 'high', 'low', 'vol', 'amt','data_slot']sio.S2M(work_out_stream,keyname= keyname, tier1= tier1,tier2=tier2,w =cur_w, batch_num =batch_num,sniffer_name=sniffer_name,log_tier2 =log_tier2, redis_agent = redis_agent_host,toDoubleVarList=toDoubleVarList)

同理，也可以修改app01_PullToStep1MongoIn.py(当时app01并没有直接使用S2M方法)

...# - toDoubleVarListif len(toDoubleVarList):for _var in toDoubleVarList:q_data2[_var] = q_data2[_var].apply(float)

6 补充

如果一开始不小心将mongo的变量格式搞错了，那么可以这么修改

import pymongo
lmongo=pymongo.MongoClient(host='172.17.0.1',port=111,username='aaa',password='bbb',authSource='admin',authMechanism='SCRAM-SHA-1')lmongo.list_database_names()# 声明数据库和集合
the_collection = lmongo['MyQuantBase']['step1_mongo_in']for varname in sorted(['amt','close','data_slot','high','low','open','vol']):command_dict = {'%s' % varname :{'$exists': True, '$type': 'string'}} command_list = [{'$set':{'%s' % varname  :  { '$toDouble': '$%s' % varname }}}]res = the_collection.update_many(command_dict, command_list)print(res.acknowledged)

7 总结

• 1 本次打通了实时的分钟级取数
• 2 对ADBS的规范进行了一些强化：例如入数据的变量要确定，不可缺失。同时对透传的Worker进行了一些适配性更改。
• 3 考虑到有一些字段是要作为过滤器的，而经过Redis队列，很多变量似乎都变字符了。所以在两个ToMongo的App(01,03）增加了转浮点的部分。(在以后的ADBS升级中可以考虑将这些固化到配置文件中)