Python 全栈系列265 使用ORM、Kafka、Apscheduler实现任务的并发处理

本文主要是介绍Python 全栈系列265 使用ORM、Kafka、Apscheduler实现任务的并发处理，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

说明

这次的尝试，从框架来说是比较成功的。但是不太走运的是，有一个小的磁盘回收没有写，结果在我外出旅游的时候磁盘打满，导致任务没有按预期执行完，这点比较遗憾。

这里快速把实现的框架梳理一下，后续可以使用，以及进一步优化。

内容

1 任务数据的分发

需要处理的任务数据，先存放在了mysql的source表，处理的结果存放在result表。

首先，我为了方便使用kafka，搭建了一个kafka agent服务。这样的好处是，在任何环境下都可以使用，这对于worker来说是方便的（不再需要考虑安装kafka的环境。但是一个比较严重的问题是，增加了两次json序列化，这对于大文本处理来说还是比较影响效率的。

from Basefuncs import *import time 
import requests as req 
from pydantic import BaseModel,field_validator
import pandas as pd 
import json 
import time 
class Producer(BaseModel):servers : str raw_msg_list : list is_json : bool = True topic : str @propertydef msg_list(self):# change raw - json if self.is_json:tick1 = time.time()the_list = pd.Series(self.raw_msg_list).apply(json.dumps).to_list()print('takes %.2f for json dumps ' %(time.time() - tick1 ))return the_list else:return self.raw_msg_listfrom sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, func, Text, Index
from sqlalchemy.orm import sessionmaker,declarative_base
from datetime import datetimem7_24013_url = f"mysql+pymysql://xxx:xxx@172.17.0.1:24013/mydb"# from urllib.parse import quote_plus
# the_passed = quote_plus('!@#*')
# # 创建数据库引擎
m7_engine = create_engine(m7_24013_url)# 创建基类
Base = declarative_base()# 定义数据模型
class DocEntMap(Base):__tablename__ = 'doc_ent_map'id = Column(Integer, primary_key=True)# CompileError: (in table 'users', column 'name'): VARCHAR requires a length on dialect mysqldoc_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),)# 定义模型类
class SourceData(Base):__tablename__ = 'source_data'id = Column(Integer, primary_key=True)mid = Column(String(50))content = Column(Text)created = Column(String(50))def dict(self):data_dict = {}data_dict['doc_id'] = self.middata_dict['text'] = self.contentreturn data_dict# 创建表（如果表已经存在，这一步将忽略）Base.metadata.create_all(m7_engine)# 创建会话
Session = sessionmaker(bind=m7_engine)session = Session()# 分页查询
page = 1
page_size = 10000while True:offset = (page - 1) * page_sizeresult = session.query(SourceData).filter(~SourceData.mid.in_(session.query(DocEntMap.doc_id))).offset(offset).limit(page_size).all()if not result:breakif page % 10 ==0:print(page)resent_task_list = [x.dict() for x in result]produces = Producer(servers = 'kafkaIP:9092',raw_msg_list = resent_task_list, topic='data_round_1' )resp = req.post('http://IP:24132/send_msg/',json = produces.dict()).json()page += 1session.close()

我让deepseek解读了一下，大体上是这个意思。

这段代码实现了一个数据处理和传输的流程，主要功能如下：

1. 导入必要的库和模块：

   • from Basefuncs import *：从 Basefuncs 模块中导入所有内容（具体内容未展示）。
   • import time：导入时间模块，用于计时和延时。
   • import requests as req：导入 requests 库，用于 HTTP 请求。
   • from pydantic import BaseModel, field_validator：从 pydantic 库中导入 BaseModel 和 field_validator，用于数据验证和模型定义。
   • import pandas as pd：导入 pandas 库，用于数据处理。
   • import json：导入 json 库，用于 JSON 数据的处理。

2. 定义 Producer 类：

   • 继承自 BaseModel，定义了四个字段：servers、raw_msg_list、is_json 和 topic。
   • msg_list 是一个属性方法，用于将 raw_msg_list 转换为 JSON 格式（如果 is_json 为 True），并计算转换时间。

3. 数据库相关操作：

   • 使用 sqlalchemy 库连接到 MySQL 数据库，并定义了两个数据模型：DocEntMap 和 SourceData。
   • DocEntMap 模型定义了一个文档与实体映射的表，包含 id、doc_id、ent_list_str、mapped_list_str 和 create_time 字段。
   • SourceData 模型定义了一个源数据表，包含 id、mid、content 和 created 字段。
   • 创建数据库表（如果表不存在），并创建一个数据库会话。

4. 数据查询和处理：

   • 使用分页查询从 SourceData 表中获取数据，跳过已经处理过的数据（通过 DocEntMap.doc_id 判断）。
   • 将查询结果转换为字典列表，并创建 Producer 对象。
   • 使用 requests 库发送 POST 请求，将数据发送到指定的 URL。

5. 循环处理：

   • 使用 while 循环进行分页查询，每次查询 10000 条数据，直到没有更多数据为止。
   • 每处理 10 页数据打印一次当前页码。
   • 最后关闭数据库会话。

总结：
这段代码的主要功能是从数据库中查询未处理的数据，将其转换为 JSON 格式，并通过 HTTP 请求发送到指定的服务器。整个过程使用了多个库和模块，包括数据验证、数据处理、数据库操作和网络请求等。

---

Producer 是kafka agent的生产者的数据模型，使用pydantic定义。DocEntMap和SourceData是mysql的数据模型，使用sqlalchemy定义。这里的作用是将未处理的数据按照批次数量，逐次进行偏移后发送到队列。

2 任务数据的本地缓存

理论上，worker应该直接从kafka队列中获取消息并处理。这里，由于任务处理大文本，受序列化影响比较大，所以我采取了本地缓存的方法。

left-right模式是常用的本地文件处理模式，left存的是原始任务数据，right存的是处理之后的结果。通过文件名的比对，可以知道完成的任务。

left-right模式还是挺好用的，比较简单、直观。过去常用的方式是按照编号规律(例如对10取余)，由若干个worker获取不同的数据处理，然后存在right下面。

考虑到本次任务处理的时间较长，且数据不再存在本地，而是通过ORM存到数据库。所以本次处理时会稍作改变：Worker取到数据时，会立即往right存一个同名空文件，起到类似消息队列中ACK的作用，从而避免其他worker重复取数。

from Basefuncs import * import shortuuiddef get_shortuuid(charset=None):"""生成一个简洁的唯一标识符（短 UUID）。参数:charset (str, optional): 自定义的字符集。如果未提供，将使用默认字符集。返回:str: 生成的短 UUID。"""if charset:su = shortuuid.ShortUUID(charset=charset)return su.uuid()else:return shortuuid.uuid()import requests as req 
from pydantic import BaseModel,field_validator
import pandas as pd 
import json 
import time # group.id: 声明不同的group.id 可以重头消费
class InputConsumer(BaseModel):servers : str groupid : str = 'default01'is_commit: bool = True msg_num : int  = 3 topic : str is_json : bool = True the_consumer = InputConsumer(servers = 'Kafka IP:9092', msg_num =100, topic='data_round_1',groupid='test02')
# the_consumer = InputConsumer(servers = '127.0.0.1:9092', msg_num =100000, topic='mytest200')
import time 
tick1 = time.time()
# resp = req.post('http://127.0.0.1:8000/consume_msg/',json = the_consumer.dict()).json()
resp = req.post('http://172.17.0.1:24132/consume_msg/',json = the_consumer.dict()).json()
print(resp[0])
tick2 = time.time()left_path = './left_v3/'to_pickle(resp, get_shortuuid(),left_path)

这段代码主要涉及以下几个部分：

1. 导入模块和函数：

   • from Basefuncs import *：从 Basefuncs 模块中导入所有内容。这通常用于导入一些基础功能函数或工具函数。
   • import shortuuid：导入 shortuuid 模块，用于生成简洁的唯一标识符。
   • import requests as req：导入 requests 模块并将其别名设置为 req，用于发送 HTTP 请求。
   • from pydantic import BaseModel, field_validator：从 pydantic 模块中导入 BaseModel 和 field_validator，用于数据验证和模型定义。
   • import pandas as pd：导入 pandas 模块并将其别名设置为 pd，用于数据处理。
   • import json 和 import time：导入 json 和 time 模块，分别用于处理 JSON 数据和时间操作。

2. 生成短 UUID 的函数：

   def get_shortuuid(charset=None):"""生成一个简洁的唯一标识符（短 UUID）。参数:charset (str, optional): 自定义的字符集。如果未提供，将使用默认字符集。返回:str: 生成的短 UUID。"""if charset:su = shortuuid.ShortUUID(charset=charset)return su.uuid()else:return shortuuid.uuid()
   

   这个函数用于生成一个短 UUID，可以选择性地使用自定义字符集。

3. 定义输入消费者模型：

   class InputConsumer(BaseModel):servers: strgroupid: str = 'default01'is_commit: bool = Truemsg_num: int = 3topic: stris_json: bool = True
   

   这个类定义了一个输入消费者模型，包含服务器地址、消费者组 ID、是否提交、消息数量、主题和是否为 JSON 格式等字段。

4. 实例化输入消费者对象：

   the_consumer = InputConsumer(servers='KAFKA IP:9092', msg_num=100, topic='data_round_1', groupid='test02')
   

   创建一个 InputConsumer 对象，并设置相关参数。

5. 发送 HTTP 请求并处理响应：

   import time
   tick1 = time.time()
   resp = req.post('http://172.17.0.1:24132/consume_msg/', json=the_consumer.dict()).json()
   print(resp[0])
   tick2 = time.time()
   

   使用 requests 模块发送一个 POST 请求，并将 the_consumer 对象转换为 JSON 格式作为请求体。然后打印响应的第一个元素，并记录时间。

6. 保存响应数据：

   left_path = './left_v3/'
   to_pickle(resp, get_shortuuid(), left_path)
   

   将响应数据保存到指定路径，使用 get_shortuuid 函数生成文件名。

总结：
这段代码主要用于生成短 UUID、定义数据模型、发送 HTTP 请求并处理响应，最后将响应数据保存到本地。

---

上面，其实是将消费者取数部分的功能独立了出来。InputConsumer是Kafka Agent消费者的数据模型，消费者通过Kafka Agent取到一批数，然后用get_shortuuid生成一个临时的不重复文件名来指代这个任务。

这样做，有点类似pre_fetch的概念，提前把数据取到，完成序列化(存为pickle)，节约了后续处理步骤的时间。数据的获取和处理两个步骤本来就应该分开。

3 处理及存储

from Basefuncs import * left_path = './left_v3/'
right_path = './right_v3/'left_files = list_file_names_without_extension(left_path)
right_files = list_file_names_without_extension(right_path)gap_files =  left_files - right_files
gap_flist = list(gap_files)
gap_file_list = sorted(list(gap_flist))# 随机挑选一个任务
some_task_file = np.random.choice(gap_file_list)
print(some_task_file)# placeholder
to_pickle('', some_task_file, right_path)some_task_data = from_pickle(some_task_file, left_path)class ListBatchIterator:def __init__(self, some_list, batch_size):self.some_list = some_listself.batch_size = batch_size@staticmethoddef slice_list_by_batch(list_length, batch_num):batch_list =list(range(0, list_length +batch_num , batch_num))res_list = []for i in range(len(batch_list)-1):res_list.append((batch_list[i],batch_list[i+1]))return res_listdef __iter__(self):the_slice_list = self.slice_list_by_batch(len(self.some_list), self.batch_size)for the_slice  in the_slice_list:yield self.some_list[the_slice[0]:the_slice[1]]lb = ListBatchIterator(some_task_data, 1)
res_list = []for small_list in lb:input_dict = {}input_dict['data_list'] = small_listserver_url = 'http://172.17.0.1:8001/ent_mapping/'resp1 = req.post(server_url, json=input_dict).json()res_list += resp1res_file_list2 = [x for x in res_list if x !='detail']# ----------------------  结果存库
from pydantic import BaseModel,field_validator
class DocEnt(BaseModel):doc_id : strent_list : list maaped_ent: list @propertydef ent_list_str(self):return ','.join(self.ent_list)@propertydef mapped_list_str(self):return ','.join(self.maaped_ent)def dict(self):data_dict = {}data_dict['doc_id'] = self.doc_iddata_dict['ent_list_str'] = self.ent_list_strdata_dict['mapped_list_str'] = self.mapped_list_strreturn data_dictfrom typing import Listclass DocEnt_list(BaseModel):data_list: List[DocEnt]doc_ent_list = DocEnt_list(data_list=res_file_list2)result = []
for x in doc_ent_list.data_list:try:result.append(x.dict())except:passfrom sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, func, Text, Index
from sqlalchemy.orm import sessionmaker,declarative_base
from datetime import datetimem7_24013_url = f"mysql+pymysql://xxx:xxx@172.17.0.1:24013/mydb"# from urllib.parse import quote_plus
# the_passed = quote_plus('!@#*')
# # 创建数据库引擎
m7_engine = create_engine(m7_24013_url)# 创建基类
Base = declarative_base()# 定义数据模型
class DocEntMap(Base):__tablename__ = 'doc_ent_map'id = Column(Integer, primary_key=True)# CompileError: (in table 'users', column 'name'): VARCHAR requires a length on dialect mysqldoc_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)with Session() as session:for tem_result in result:# print(tem_result)try:session.add(DocEntMap(**tem_result))session.commit()except:pass

这段代码主要涉及以下几个部分：

1. 导入模块和函数：

   • from Basefuncs import *：从 Basefuncs 模块中导入所有内容。这通常用于导入一些基础功能函数或工具函数。
   • import numpy as np：导入 numpy 模块并将其别名设置为 np，用于数值计算。
   • from pydantic import BaseModel, field_validator：从 pydantic 模块中导入 BaseModel 和 field_validator，用于数据验证和模型定义。
   • from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, func, Text, Index：从 sqlalchemy 模块中导入用于数据库操作的各种类和函数。
   • from sqlalchemy.orm import sessionmaker, declarative_base：从 sqlalchemy.orm 模块中导入用于 ORM 操作的类和函数。
   • from datetime import datetime：导入 datetime 模块，用于处理日期和时间。

2. 列出文件名并找出差异：

   left_path = './left_v3/'
   right_path = './right_v3/'left_files = list_file_names_without_extension(left_path)
   right_files = list_file_names_without_extension(right_path)gap_files = left_files - right_files
   gap_flist = list(gap_files)
   gap_file_list = sorted(list(gap_flist))
   

   这段代码列出 left_path 和 right_path 目录下的文件名（不包括扩展名），并找出 left_path 中有但 right_path 中没有的文件名。

3. 随机选择一个任务文件：

   some_task_file = np.random.choice(gap_file_list)
   print(some_task_file)
   

   从差异文件列表中随机选择一个文件名，并打印出来。

4. 保存占位符文件：

   to_pickle('', some_task_file, right_path)
   

   将一个空字符串保存到 right_path 目录下，文件名为 some_task_file。

5. 加载任务数据：

   some_task_data = from_pickle(some_task_file, left_path)
   

   从 left_path 目录下加载 some_task_file 文件的数据。

6. 定义 ListBatchIterator 类：

   class ListBatchIterator:def __init__(self, some_list, batch_size):self.some_list = some_listself.batch_size = batch_size@staticmethoddef slice_list_by_batch(list_length, batch_num):batch_list = list(range(0, list_length + batch_num, batch_num))res_list = []for i in range(len(batch_list) - 1):res_list.append((batch_list[i], batch_list[i + 1]))return res_listdef __iter__(self):the_slice_list = self.slice_list_by_batch(len(self.some_list), self.batch_size)for the_slice in the_slice_list:yield self.some_list[the_slice[0]:the_slice[1]]
   

   这个类用于将列表按批次分割，并提供迭代器功能。

7. 发送 HTTP 请求并处理响应：

   lb = ListBatchIterator(some_task_data, 1)
   res_list = []for small_list in lb:input_dict = {}input_dict['data_list'] = small_listserver_url = 'http://172.17.0.1:8001/ent_mapping/'resp1 = req.post(server_url, json=input_dict).json()res_list += resp1res_file_list2 = [x for x in res_list if x != 'detail']
   

   使用 ListBatchIterator 类按批次处理数据，并发送 HTTP 请求获取响应，然后将响应数据存储到 res_list 中。

8. 定义数据模型：

   class DocEnt(BaseModel):doc_id: strent_list: listmaaped_ent: list@propertydef ent_list_str(self):return ','.join(self.ent_list)@propertydef mapped_list_str(self):return ','.join(self.maaped_ent)def dict(self):data_dict = {}data_dict['doc_id'] = self.doc_iddata_dict['ent_list_str'] = self.ent_list_strdata_dict['mapped_list_str'] = self.mapped_list_strreturn data_dictclass DocEnt_list(BaseModel):data_list: List[DocEnt]
   

   定义 DocEnt 和 DocEnt_list 类，用于数据模型和验证。

9. 处理结果并存入数据库：

   doc_ent_list = DocEnt_list(data_list=res_file_list2)result = []
   for x in doc_ent_list.data_list:try:result.append(x.dict())except:passm7_24013_url = f"mysql+pymysql://xxx:xxx@172.17.0.1:24013/mydb"
   m7_engine = create_engine(m7_24013_url)Base = declarative_base()class DocEntMap(Base):__tablename__ = 'doc_ent_map'id = Column(Integer, primary_key=True)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)with Session() as session:for tem_result in result:try:session.add(DocEntMap(**tem_result))session.commit()except:pass
   

   将处理后的结果存入 MySQL 数据库中。

总结：
这段代码主要用于处理文件名差异、随机选择任务文件、加载任务数据、按批次处理数据、发送 HTTP 请求、处理响应、定义数据模型，并将结果存入数据库。

---

总体上，处理过程获取任务数据，然后ACK(在right立即创建文件，下一次worker不会再取到该任务)。然后将数据发起web请求进行处理(这里又涉及到序列化和反序列化),结果通过ORM存到mysql。

4 调度

使用apscheduler调度

from datetime import datetime
import os 
from apscheduler.schedulers.blocking import BlockingSchedulerdef exe_sh(cmd = None):os.system(cmd)# 后台启动命令 nohup python3 aps_v2.py >/dev/null 2>&1 &if __name__ == '__main__':sche1 = BlockingScheduler()# sche1.add_job(exe_sh,'interval', seconds=1, kwargs ={'cmd':'python3 ./main_handler/main.py'})sche1.add_job(exe_sh,'interval', seconds=1, kwargs ={'cmd':'python3 sniffer_v2.py'},max_instances=1,coalesce=True)sche1.add_job(exe_sh,'interval', seconds=1, kwargs ={'cmd':'python3 random_worker_v2.py'},max_instances=6,coalesce=True)print('[S] starting inteverl')sche1.start()

这段代码主要涉及以下几个部分：

1. 导入模块：

   • from datetime import datetime：导入 datetime 模块，用于处理日期和时间。
   • import os：导入 os 模块，用于与操作系统进行交互。
   • from apscheduler.schedulers.blocking import BlockingScheduler：从 apscheduler 模块中导入 BlockingScheduler，用于创建和调度任务。

2. 定义 exe_sh 函数：

   def exe_sh(cmd=None):os.system(cmd)
   

   这个函数用于执行传入的 shell 命令。

3. 主程序：

   if __name__ == '__main__':sche1 = BlockingScheduler()sche1.add_job(exe_sh, 'interval', seconds=1, kwargs={'cmd': 'python3 sniffer_v2.py'},max_instances=1, coalesce=True)sche1.add_job(exe_sh, 'interval', seconds=1, kwargs={'cmd': 'python3 random_worker_v2.py'},max_instances=6, coalesce=True)print('[S] starting interval')sche1.start()
   

   这段代码的主要功能是创建一个 BlockingScheduler 实例，并添加两个定时任务：

   • 第一个任务每秒执行一次 python3 sniffer_v2.py 命令，最多允许一个实例运行，并且合并执行。
   • 第二个任务每秒执行一次 python3 random_worker_v2.py 命令，最多允许六个实例运行，并且合并执行。

   max_instances 参数指定允许同时运行的最大实例数，coalesce 参数指定当多个任务触发时是否合并为一个任务执行。

4. 后台启动命令：

   # 后台启动命令 nohup python3 aps_v2.py >/dev/null 2>&1 &
   

   这是一个注释，说明如何后台启动这个脚本。nohup 命令用于在后台运行程序，并将输出重定向到 /dev/null，2>&1 将标准错误输出重定向到标准输出。

总结：
这段代码使用 apscheduler 库创建了一个阻塞式调度器，并添加了两个定时任务，分别每秒执行不同的 Python 脚本。这些任务可以在后台运行，不会阻塞当前终端。

---

这里用aps,效果类似于使用threading

def worker(thread_id, some_data):print(f"Thread {thread_id} started")input_dict = {'data_list': [some_data]}  # 将单个数据条目包装在列表中# 如果 resp 是一个列表，直接使用它来创建 DocEnt_listif isinstance(resp, list):doc_ent_list = DocEnt_list(data_list=resp)result = [x.dict() for x in doc_ent_list.data_list]else:print(f"Thread {thread_id} received invalid data: {resp}")return []print(f"Thread {thread_id} finished with result: {result}")return result# 使用 ThreadPoolExecutor 来管理线程并获取结果
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:# 提交任务并获取 Future 对象futures = [executor.submit(worker, i, some_data) for i, some_data in enumerate(resp)]# 获取结果results = [future.result() for future in concurrent.futures.as_completed(futures)]print("All threads finished")
print("Results:", results)

5 总结

这种实现方式还是比较直观的，通过类似DBeaver之类的终端连接上数据库后，我们可以清楚的看到源数据，也可以看到随时间推移，不断的有处理好的数据存在结果数据库。

有些地方可以稍微关注的是：

• 1 ORM可以挂clickhouse， 这样存储所占的空间更小，在后续的提取、统计数据时更快。但需要验证的是clickhouse的where in 效率。
• 2 ORM是可以批量存数的，这个是稍微靠后才发现的。
• 3 FastAPI + APScheduler + Celery + Dash 调度。 目前用了Flask-APScheduler和Flask-Celery，但感觉不是很趁手。之后把这几个组件分开来研究一下，再重新组装。特别是可视化和控制这块，需要结合ORM增强。

这篇关于Python 全栈系列265 使用ORM、Kafka、Apscheduler实现任务的并发处理的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---