Python 全栈系列264 使用kafka进行并发处理

说明

暂时考虑的场景是单条数据处理特别复杂和耗时的场景。

场景如下：

要对一篇文档进行实体处理，然后再对实体进行匹配。在这个过程当中，涉及到了好几部分服务：

• 1 实体识别服务
• 2 数据库查询服务
• 3 es查询服务

整个处理包成了服务，在单线程处理增量的时候非常正常，但尝试进行并行调用的时候出现了问题。每次报错的时候都是显示

Connection Reset By Peer

,感觉像是服务端连接的问题。由于每一部分都可能是瓶颈，我没(时间)法准确定位问题所在，很有可能是同时起了5个实体识别，GPU的抢占导致的问题(负载经常在100%)。

所以，这个事有两个启示：

• 1 对于One处理的设计，是否可以保存中间关键步骤的跟踪？
• 2 系统资源瓶颈。服务资源是瓶颈(GPU、网络、数据库IO),如果目标是瓶颈资源(Server模式)，很容易出现失败。反过来，如果从瓶颈资源出发,尽力而为(Worker)反而会有更高的资源利用率。但是，Server模式是面向消费者的，而Worker模式则是面向生产者的，如果我们要把工作交出去，还是应该采用Server模式。

所以要解决的问题是：在保证逻辑正确的情况下，且有大量miss的情况下，如何尽快的完成业务上的任务。

内容

在数据处理架构中,可靠性与效率也是一对钳制指标，这类型指标混在一起是不行的，必须要分开。所以在机器学习里有：

• 1 精确率与召回率。是分开来研究的，当然也有最终融合的指标(F1Score)
• 2 TCP/IP。

TCP/IP 协议栈中的两个核心协议是：

TCP（传输控制协议）：提供可靠的、面向连接的通信服务，确保数据包按顺序到达且无错误。

IP（互联网协议）：负责将数据包从源地址传输到目的地址，但不保证传输的可靠性。

在我的业务场景下，

TCP可以视为数据库+比对，而IP则视为队列+处理

。

(TCP模式)数据库的设计目的就是为了可靠、长期地保存大量的数据，我们把任务、结果以及里程碑节点(中间过程)保存在数据库中是合适的。现在假设只有两个节点：任务和结果。（虽然我发现，在有明显瓶颈的地方是实体识别，这里应该独立一个里程碑节点出来）

(IP模式)队列的设计目的则是缓冲和分发。缓冲是解决生产者的困扰，这样不需要知道机器的能力是多少，把要做的任务发完就好了。然后还可以解决“众包”问题，通过多个worker进行任务分摊，尽可能快的执行任务。在这个过程中，必然会产生大量的不确定问题，导致worker处理、交付失败的。这也是快所要付出的代价。

通过数据库 + 队列的组合，就可以做到既快又好。

1 任务数据入库

简单起见，暂时还是采用mysql。

先将原始数据灌到source表。

之前用离线方式跑了一部分数据，将这部分数据搬到 result表。

right_files = list_file_names_without_extension(right_path)
res_file_list =[]
get_right_file_names =[]for the_file inlist(right_files):
    the_tem_data = from_pickle(the_file,'./right/')
    get_right_file_names.append(the_file)
    res_file_list.append(the_tem_data)

res_file_list1 = flatten_list(res_file_list)# 过滤掉失败的
res_file_list2 =[x for x in res_file_list1 if x !='detail']
right_res_df = pd.DataFrame(res_file_list2)# 引入与数据库表规范对接的数据模型from pydantic import BaseModel,field_validator
classDocEnt(BaseModel):
    doc_id :str
    ent_list :list 
    maaped_ent:list@propertydefent_list_str(self):return','.join(self.ent_list)@propertydefmapped_list_str(self):return','.join(self.maaped_ent)defdict(self):
        data_dict ={}
        data_dict['doc_id']= self.doc_id
        data_dict['ent_list_str']= self.ent_list_str
        data_dict['mapped_list_str']= self.mapped_list_str
        return data_dict

from typing import List

classDocEnt_list(BaseModel):
    data_list: List[DocEnt]# 将结果数据转为可被数据库接受的字段模式
docent_list = DocEnt_list(data_list = right_res_df.to_dict(orient='records'))
docent_list1 =[x.dict()for x in docent_list.data_list]

将合法的数据结果与ORM对接,先引入数据模型。

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, func, Text, Index
from sqlalchemy.orm import sessionmaker,declarative_base
from datetime import datetime

m7_24013_url =f"mysql+pymysql://xxx:[email protected]:24013/mydb"# from urllib.parse import quote_plus# the_passed = quote_plus('!@#*')# # 创建数据库引擎
m7_engine = create_engine(m7_24013_url)# 创建基类
Base = declarative_base()# 定义数据模型classDocEntMap(Base):
    __tablename__ ='doc_ent_map'id= Column(Integer, primary_key=True)# CompileError: (in table 'users', column 'name'): VARCHAR requires a length on dialect mysql
    doc_id = Column(String(50))
    ent_list_str = Column(Text)
    mapped_list_str = Column(Text)
    create_time = Column(DateTime, default=lambda: datetime.now())# 创建索引
    __table_args__ =(
        Index('idx_doc_id', doc_id),
        Index('idx_create_time', create_time),)

Base.metadata.create_all(m7_engine)# 创建会话
Session = sessionmaker(bind=m7_engine)

分批次存储数据

ent_map_lb = ListBatchIterator(docent_list1,1000)import tqdm
with Session()as session:for i,some_list in tqdm.tqdm(enumerate(ent_map_lb)):
        test_list =[DocEntMap(**x)for x in some_list]# 一次性添加到会话中
        session.add_all(test_list)# 提交会话
        session.commit()

因为是mysql，我按照每批1000来操作，每秒能存2批，这个速度也能接受了。

2 比较差集

方式一：mysql不支持

SELECT id FROM table1
EXCEPTDISTINCTSELECT id FROM table2;

方式2：left join

-- 查找在 table1 中存在但在 table2 中不存在的 idSELECT t1.id FROM table1 t1
LEFTJOIN table2 t2 ON t1.id = t2.id
WHERE t2.id ISNULL;

我使用Sqlalchmey进行比较并获取数据，稍慢，但方法简单

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, func, Text, Index
from sqlalchemy.orm import sessionmaker,declarative_base
from datetime import datetime

m7_24013_url =f"mysql+pymysql://xxx:[email protected]:24013/mydb"# from urllib.parse import quote_plus# the_passed = quote_plus('!@#*')# # 创建数据库引擎
m7_engine = create_engine(m7_24013_url)# 创建基类
Base = declarative_base()# 定义数据模型classDocEntMap(Base):
    __tablename__ ='doc_ent_map'id= Column(Integer, primary_key=True)# CompileError: (in table 'users', column 'name'): VARCHAR requires a length on dialect mysql
    doc_id = Column(String(50))
    ent_list_str = Column(Text)
    mapped_list_str = Column(Text)
    create_time = Column(DateTime, default=lambda: datetime.now())# 创建索引
    __table_args__ =(
        Index('idx_doc_id', doc_id),
        Index('idx_create_time', create_time),)# 定义模型类classSourceData(Base):
    __tablename__ ='source_data'id= Column(Integer, primary_key=True)
    mid = Column(String(50))
    content = Column(Text)
    created = Column(String(50))defdict(self):
        data_dict ={}
        data_dict['doc_id']= self.mid
        data_dict['text']= self.content
        return data_dict

# 创建表（如果表已经存在，这一步将忽略）
Base.metadata.create_all(m7_engine)# 创建会话
Session = sessionmaker(bind=m7_engine)# 分页查询
page =1
page_size =1000whileTrue:
    offset =(page -1)* page_size
    result = session.query(SourceData).filter(~SourceData.mid.in_(session.query(DocEntMap.doc_id))).offset(offset).limit(page_size).all()ifnot result:breakif page %10==0:print(page)
    resent_task_list =[x.dict()for x in result]
    produces = Producer(servers ='KAFKASERVER',raw_msg_list = resent_task_list, topic='the_topic')
    resp = req.post('http://agent:port/send_msg/',json = produces.dict()).json()
    page +=1

翻页到后面还是慢的，不过确实比较简单，省事。

如果使用clickhouse,选取列的速度还是非常快的。要做好索引之后再取数应该效率会比较高。后续再看吧，我现在都倾向先用lazy版的。

sql="""
    SELECT mid 
    FROM source
    EXCEPT
    SELECT doc_id 
    FROM target
;
"""data=chc._exe_sql(sql)

3 kafka 消费

消费也是通过kafka agent来做的，但是比我之前在本机跑慢。我猜是因为worker在处理完任务，请求下一个时，无论多快，都要进行序列化。而且因为通过agent进行，消费者需要对数据进行两次序列化，这个还是会比较耗时的。

我的想法是，通过一个专门的数据拉取程序，事先将数据从kafka上拉下来，以short_uuid命名，然后存在本地的left。然后本地的worker从left中取数。

为什么要大费周章从kafka取，而不是直接从数据库取？

主要目的是为了更好的分发。例如此时又要加另外两台机器协同处理，难道我还要在手工分配数据吗？

纯粹取数的程序，消费速度一定比处理程序快多了，这样就避免了每次要处理时才进行序列化。