YOLO12模型在MySQL数据库中的结果存储与查询优化如果你正在用YOLO12做目标检测每天处理成千上万张图片检测结果越来越多怎么存、怎么查就成了大问题。直接存成文件找起来麻烦分析起来更麻烦。用数据库怎么设计表结构才能既存得下又查得快今天咱们就来聊聊怎么把YOLO12的检测结果高效地存进MySQL还能让查询飞起来。我做了10多年AI项目踩过不少坑也总结了一些实用的经验分享给你。1. 为什么要把检测结果存进数据库你可能觉得检测结果不就是一些坐标和类别吗存成JSON或者CSV文件不就行了刚开始确实可以但项目做大了就会发现问题。我去年帮一个安防公司做项目他们用YOLO12做实时监控每天产生几十万条检测记录。刚开始他们用文件存储结果一个月后想查某个摄像头某段时间的检测记录得遍历几十个文件每次查询都要好几分钟。后来我们改成了数据库存储同样的查询现在只要几秒钟。数据库存储有几个明显的好处查询方便想查什么就查什么按时间、按类别、按摄像头随便组合统计分析可以轻松做各种统计比如“今天上午9点到10点3号摄像头检测到多少人”数据关联可以把检测结果和其他业务数据关联起来比如“检测到异常时自动触发报警”长期存储数据不会丢失方便回溯和分析2. 数据库表结构设计怎么存才合理设计表结构是个技术活设计得好查询快如闪电设计得不好查询慢如蜗牛。下面是我在实际项目中总结出来的设计思路。2.1 核心表设计先来看最核心的检测结果表。这个表要存每次检测的具体信息CREATE TABLE detection_results ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY COMMENT 主键ID, image_id VARCHAR(100) NOT NULL COMMENT 图片或视频帧的唯一标识, camera_id VARCHAR(50) COMMENT 摄像头编号, detection_time DATETIME NOT NULL COMMENT 检测时间, model_version VARCHAR(20) DEFAULT yolo12 COMMENT 模型版本, -- 检测对象信息 class_id INT NOT NULL COMMENT 类别ID, class_name VARCHAR(50) NOT NULL COMMENT 类别名称, confidence DECIMAL(5,4) NOT NULL COMMENT 置信度0.0000-1.0000, -- 边界框坐标归一化到0-1 x_center DECIMAL(6,5) NOT NULL COMMENT 中心点x坐标, y_center DECIMAL(6,5) NOT NULL COMMENT 中心点y坐标, width DECIMAL(6,5) NOT NULL COMMENT 边界框宽度, height DECIMAL(6,5) NOT NULL COMMENT 边界框高度, -- 原始像素坐标可选根据需求存储 pixel_x INT COMMENT 原始像素x坐标, pixel_y INT COMMENT 原始像素y坐标, pixel_width INT COMMENT 原始像素宽度, pixel_height INT COMMENT 原始像素高度, -- 其他信息 image_width INT COMMENT 图片宽度, image_height INT COMMENT 图片高度, processing_time_ms INT COMMENT 处理耗时毫秒, -- 索引字段 INDEX idx_detection_time (detection_time), INDEX idx_camera_time (camera_id, detection_time), INDEX idx_class_time (class_id, detection_time), INDEX idx_confidence (confidence) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENTYOLO12检测结果表;这个设计有几个关键点主键用自增ID查询快插入也快坐标存归一化值这样无论原始图片多大坐标都在0-1之间方便后续处理分开存类别ID和名称ID用于查询名称用于显示记录图片尺寸如果需要还原原始坐标有这个信息就行2.2 分类信息表如果类别比较多或者类别信息会变化最好单独建个表CREATE TABLE detection_classes ( class_id INT NOT NULL PRIMARY KEY COMMENT 类别ID, class_name VARCHAR(50) NOT NULL COMMENT 类别名称, display_name VARCHAR(50) COMMENT 显示名称, is_active TINYINT DEFAULT 1 COMMENT 是否启用, created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间, UNIQUE KEY uk_class_name (class_name), INDEX idx_is_active (is_active) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT检测类别表; -- 插入一些常见类别 INSERT INTO detection_classes (class_id, class_name, display_name) VALUES (0, person, 人), (1, bicycle, 自行车), (2, car, 汽车), (3, motorcycle, 摩托车), -- ... 其他YOLO12支持的类别 (56, chair, 椅子), (60, dining table, 餐桌);2.3 统计表可选如果经常需要做统计可以建个统计表每天或每小时更新一次CREATE TABLE detection_statistics ( stat_date DATE NOT NULL COMMENT 统计日期, camera_id VARCHAR(50) COMMENT 摄像头编号, class_id INT NOT NULL COMMENT 类别ID, detection_count INT DEFAULT 0 COMMENT 检测次数, avg_confidence DECIMAL(5,4) COMMENT 平均置信度, peak_hour TIME COMMENT 检测高峰时段, PRIMARY KEY (stat_date, camera_id, class_id), INDEX idx_date_camera (stat_date, camera_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT检测统计表;3. 从YOLO12到MySQL代码怎么写表设计好了接下来就是怎么把YOLO12的检测结果存进去。我用Python写了个简单的例子你可以参考一下。3.1 安装必要的库# 安装MySQL连接库 # pip install mysql-connector-python # 或者 # pip install pymysql # 安装YOLO12 # pip install ultralytics3.2 连接数据库的工具类import mysql.connector from mysql.connector import Error import json from datetime import datetime class DetectionDB: def __init__(self, hostlocalhost, databaseyolo_detection, userroot, passwordyour_password): self.host host self.database database self.user user self.password password self.connection None def connect(self): 连接到MySQL数据库 try: self.connection mysql.connector.connect( hostself.host, databaseself.database, userself.user, passwordself.password ) if self.connection.is_connected(): print(成功连接到MySQL数据库) return True except Error as e: print(f连接数据库时出错: {e}) return False def disconnect(self): 断开数据库连接 if self.connection and self.connection.is_connected(): self.connection.close() print(数据库连接已关闭) def save_detection_result(self, detection_data): 保存单条检测结果 if not self.connection: print(数据库未连接) return False try: cursor self.connection.cursor() # 构建插入SQL sql INSERT INTO detection_results ( image_id, camera_id, detection_time, model_version, class_id, class_name, confidence, x_center, y_center, width, height, pixel_x, pixel_y, pixel_width, pixel_height, image_width, image_height, processing_time_ms ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) # 执行插入 cursor.execute(sql, ( detection_data[image_id], detection_data.get(camera_id), detection_data[detection_time], detection_data.get(model_version, yolo12), detection_data[class_id], detection_data[class_name], float(detection_data[confidence]), float(detection_data[x_center]), float(detection_data[y_center]), float(detection_data[width]), float(detection_data[height]), detection_data.get(pixel_x), detection_data.get(pixel_y), detection_data.get(pixel_width), detection_data.get(pixel_height), detection_data.get(image_width), detection_data.get(image_height), detection_data.get(processing_time_ms) )) self.connection.commit() print(f检测结果已保存ID: {cursor.lastrowid}) cursor.close() return cursor.lastrowid except Error as e: print(f保存检测结果时出错: {e}) return False def batch_save_results(self, results_list): 批量保存检测结果 if not self.connection: print(数据库未连接) return False try: cursor self.connection.cursor() sql INSERT INTO detection_results ( image_id, camera_id, detection_time, model_version, class_id, class_name, confidence, x_center, y_center, width, height, pixel_x, pixel_y, pixel_width, pixel_height, image_width, image_height, processing_time_ms ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) # 准备数据 data_to_insert [] for detection in results_list: data_to_insert.append(( detection[image_id], detection.get(camera_id), detection[detection_time], detection.get(model_version, yolo12), detection[class_id], detection[class_name], float(detection[confidence]), float(detection[x_center]), float(detection[y_center]), float(detection[width]), float(detection[height]), detection.get(pixel_x), detection.get(pixel_y), detection.get(pixel_width), detection.get(pixel_height), detection.get(image_width), detection.get(image_height), detection.get(processing_time_ms) )) # 批量插入 cursor.executemany(sql, data_to_insert) self.connection.commit() print(f批量保存完成共插入 {cursor.rowcount} 条记录) cursor.close() return True except Error as e: print(f批量保存时出错: {e}) return False3.3 YOLO12检测结果处理from ultralytics import YOLO import cv2 from datetime import datetime import time class YOLO12Detector: def __init__(self, model_pathyolo12n.pt): 初始化YOLO12模型 print(正在加载YOLO12模型...) self.model YOLO(model_path) print(模型加载完成) # 连接到数据库 self.db DetectionDB() self.db.connect() def process_image(self, image_path, camera_idNone): 处理单张图片并保存结果到数据库 # 读取图片 image cv2.imread(image_path) if image is None: print(f无法读取图片: {image_path}) return image_height, image_width image.shape[:2] image_id fimg_{int(time.time())}_{hash(image_path) % 10000} # 使用YOLO12进行检测 start_time time.time() results self.model(image) processing_time int((time.time() - start_time) * 1000) # 毫秒 detection_time datetime.now() detection_results [] # 处理检测结果 for result in results: boxes result.boxes if boxes is not None: for box in boxes: # 获取检测信息 class_id int(box.cls[0]) class_name self.model.names[class_id] confidence float(box.conf[0]) # 获取边界框坐标归一化 x_center, y_center, width, height box.xywhn[0].tolist() # 计算原始像素坐标 pixel_x int(x_center * image_width - width * image_width / 2) pixel_y int(y_center * image_height - height * image_height / 2) pixel_width int(width * image_width) pixel_height int(height * image_height) # 构建检测数据 detection_data { image_id: image_id, camera_id: camera_id, detection_time: detection_time, model_version: yolo12, class_id: class_id, class_name: class_name, confidence: confidence, x_center: x_center, y_center: y_center, width: width, height: height, pixel_x: pixel_x, pixel_y: pixel_y, pixel_width: pixel_width, pixel_height: pixel_height, image_width: image_width, image_height: image_height, processing_time_ms: processing_time } detection_results.append(detection_data) # 保存到数据库 if detection_results: self.db.batch_save_results(detection_results) print(f图片 {image_path} 检测完成发现 {len(detection_results)} 个目标) else: print(f图片 {image_path} 未检测到目标) return detection_results def process_video_frame(self, frame, frame_id, camera_idNone): 处理视频帧 if frame is None: return [] image_height, image_width frame.shape[:2] image_id fframe_{camera_id}_{frame_id} if camera_id else fframe_{frame_id} # 使用YOLO12进行检测 start_time time.time() results self.model(frame) processing_time int((time.time() - start_time) * 1000) detection_time datetime.now() detection_results [] for result in results: boxes result.boxes if boxes is not None: for box in boxes: class_id int(box.cls[0]) class_name self.model.names[class_id] confidence float(box.conf[0]) x_center, y_center, width, height box.xywhn[0].tolist() detection_data { image_id: image_id, camera_id: camera_id, detection_time: detection_time, model_version: yolo12, class_id: class_id, class_name: class_name, confidence: confidence, x_center: x_center, y_center: y_center, width: width, height: height, image_width: image_width, image_height: image_height, processing_time_ms: processing_time } detection_results.append(detection_data) # 批量保存 if detection_results: self.db.batch_save_results(detection_results) return detection_results def close(self): 关闭资源 self.db.disconnect()3.4 使用示例# 使用示例 def main(): # 初始化检测器 detector YOLO12Detector(yolo12n.pt) try: # 处理单张图片 print(处理单张图片示例...) results detector.process_image(path/to/your/image.jpg, camera_idcam001) # 处理多张图片 image_paths [image1.jpg, image2.jpg, image3.jpg] for img_path in image_paths: detector.process_image(img_path, camera_idcam001) # 处理视频简化示例 print(\n处理视频示例...) import cv2 video_path path/to/your/video.mp4 cap cv2.VideoCapture(video_path) frame_id 0 while cap.isOpened(): ret, frame cap.read() if not ret: break # 每10帧处理一次根据实际需求调整 if frame_id % 10 0: detector.process_video_frame(frame, frame_id, camera_idcam002) frame_id 1 # 简单控制处理100帧后停止 if frame_id 100: break cap.release() finally: # 关闭资源 detector.close() if __name__ __main__: main()4. 查询优化让数据检索飞起来数据存进去了怎么查得快才是关键。下面分享几个实用的查询优化技巧。4.1 基础查询示例class DetectionQuery: def __init__(self, db_config): self.db DetectionDB(**db_config) self.db.connect() def get_recent_detections(self, limit100): 获取最近的检测记录 query SELECT dr.*, dc.display_name FROM detection_results dr LEFT JOIN detection_classes dc ON dr.class_id dc.class_id ORDER BY dr.detection_time DESC LIMIT %s cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (limit,)) results cursor.fetchall() cursor.close() return results def get_detections_by_class(self, class_name, start_timeNone, end_timeNone): 按类别查询检测记录 query SELECT * FROM detection_results WHERE class_name %s params [class_name] if start_time: query AND detection_time %s params.append(start_time) if end_time: query AND detection_time %s params.append(end_time) query ORDER BY detection_time DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, params) results cursor.fetchall() cursor.close() return results def get_detections_by_camera(self, camera_id, hours24): 查询指定摄像头最近N小时的检测记录 query SELECT class_name, COUNT(*) as count, AVG(confidence) as avg_confidence, DATE_FORMAT(detection_time, %%Y-%%m-%%d %%H:00:00) as hour_group FROM detection_results WHERE camera_id %s AND detection_time DATE_SUB(NOW(), INTERVAL %s HOUR) GROUP BY class_name, hour_group ORDER BY hour_group DESC, count DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (camera_id, hours)) results cursor.fetchall() cursor.close() return results def get_statistics_by_date(self, dateNone): 按日期统计检测结果 if date is None: date datetime.now().date() query SELECT class_name, COUNT(*) as total_count, AVG(confidence) as avg_confidence, MIN(confidence) as min_confidence, MAX(confidence) as max_confidence, COUNT(DISTINCT camera_id) as camera_count FROM detection_results WHERE DATE(detection_time) %s GROUP BY class_name ORDER BY total_count DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (date,)) results cursor.fetchall() cursor.close() return results4.2 高级查询技巧当数据量大的时候这些查询技巧能帮你节省很多时间def get_high_confidence_detections(self, min_confidence0.8, limit1000): 查询高置信度的检测记录使用索引 # 注意我们在表上创建了 idx_confidence 索引 query SELECT /* INDEX(detection_results idx_confidence) */ image_id, camera_id, detection_time, class_name, confidence, x_center, y_center FROM detection_results WHERE confidence %s ORDER BY confidence DESC, detection_time DESC LIMIT %s cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (min_confidence, limit)) return cursor.fetchall() def get_detections_in_area(self, x_min, x_max, y_min, y_max, start_timeNone, end_timeNone): 查询指定区域内的检测记录 # 这个查询可能比较慢需要根据实际情况优化 query SELECT * FROM detection_results WHERE x_center BETWEEN %s AND %s AND y_center BETWEEN %s AND %s AND confidence 0.5 params [x_min, x_max, y_min, y_max] if start_time: query AND detection_time %s params.append(start_time) if end_time: query AND detection_time %s params.append(end_time) query ORDER BY detection_time DESC LIMIT 1000 cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, params) return cursor.fetchall()4.3 数据库性能优化建议合理使用索引在经常查询的字段上创建索引比如 detection_time、camera_id、class_id复合索引要注意顺序比如 idx_camera_time (camera_id, detection_time)不要过度索引索引会影响插入速度分区表 如果数据量特别大比如每天百万条可以考虑按时间分区-- 按月分区 ALTER TABLE detection_results PARTITION BY RANGE (YEAR(detection_time) * 100 MONTH(detection_time)) ( PARTITION p202501 VALUES LESS THAN (202502), PARTITION p202502 VALUES LESS THAN (202503), PARTITION p202503 VALUES LESS THAN (202504), PARTITION p_future VALUES LESS THAN MAXVALUE );定期清理旧数据-- 删除3个月前的数据 DELETE FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 3 MONTH); -- 或者移到历史表 INSERT INTO detection_results_history SELECT * FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 6 MONTH); DELETE FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 6 MONTH);查询优化技巧避免 SELECT *只选择需要的字段使用 EXPLAIN 分析查询计划对于复杂查询考虑使用物化视图5. 实际应用场景5.1 安防监控系统在安防监控中我们可能需要实时查询某个区域的人员密度def get_person_density_by_area(self, camera_id, hours1): 查询指定摄像头最近1小时的人员密度变化 query SELECT DATE_FORMAT(detection_time, %%Y-%%m-%%d %%H:%%i:00) as minute_time, COUNT(*) as person_count, AVG(confidence) as avg_confidence FROM detection_results WHERE camera_id %s AND class_name person AND detection_time DATE_SUB(NOW(), INTERVAL %s HOUR) GROUP BY minute_time ORDER BY minute_time cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (camera_id, hours)) results cursor.fetchall() cursor.close() # 计算密度变化 density_data [] for row in results: density_data.append({ time: row[minute_time], count: row[person_count], density: row[person_count] / 100 # 假设每100像素一个人 }) return density_data5.2 零售客流分析在零售场景中可以分析顾客在店内的行为def analyze_customer_behavior(self, store_id, date): 分析顾客行为 query SELECT camera_id, class_name, HOUR(detection_time) as hour, COUNT(*) as detection_count, AVG(confidence) as avg_confidence, -- 计算平均停留位置 AVG(x_center) as avg_x, AVG(y_center) as avg_y FROM detection_results WHERE camera_id LIKE %s AND DATE(detection_time) %s AND class_name IN (person, shopping cart, bag) GROUP BY camera_id, class_name, HOUR(detection_time) ORDER BY camera_id, hour, class_name cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (f{store_id}%, date)) results cursor.fetchall() cursor.close() # 按小时分析 hourly_analysis {} for row in results: hour row[hour] if hour not in hourly_analysis: hourly_analysis[hour] {person: 0, other: 0} if row[class_name] person: hourly_analysis[hour][person] row[detection_count] else: hourly_analysis[hour][other] row[detection_count] return { detail: results, hourly_summary: hourly_analysis }5.3 工业质检系统在工业质检中需要快速查询缺陷产品的检测记录def get_defect_statistics(self, production_line, start_date, end_date): 获取缺陷产品统计 query SELECT class_name as defect_type, COUNT(*) as defect_count, DATE(detection_time) as defect_date, camera_id as station, -- 计算缺陷率 ROUND(COUNT(*) * 100.0 / ( SELECT COUNT(*) FROM detection_results WHERE camera_id dr.camera_id AND DATE(detection_time) DATE(dr.detection_time) AND class_name ! ok ), 2) as defect_rate FROM detection_results dr WHERE camera_id LIKE %s AND detection_time BETWEEN %s AND %s AND class_name ! ok -- 假设ok表示合格品 AND confidence 0.7 GROUP BY defect_type, defect_date, station ORDER BY defect_date DESC, defect_count DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (fline{production_line}_%, start_date, end_date)) results cursor.fetchall() cursor.close() return results6. 常见问题与解决方案在实际使用中你可能会遇到这些问题6.1 数据量太大查询慢怎么办问题检测结果表有上千万条记录查询越来越慢。解决方案分区表按时间分区比如每个月一个分区归档历史数据把3个月前的数据移到历史表添加合适索引在经常查询的字段上加索引使用查询缓存对于不经常变动的统计查询可以用缓存-- 示例创建月度分区 ALTER TABLE detection_results PARTITION BY RANGE (TO_DAYS(detection_time)) ( PARTITION p202501 VALUES LESS THAN (TO_DAYS(2025-02-01)), PARTITION p202502 VALUES LESS THAN (TO_DAYS(2025-03-01)), PARTITION p202503 VALUES LESS THAN (TO_DAYS(2025-04-01)) );6.2 并发写入性能问题问题多个摄像头同时写入数据库压力大。解决方案批量插入不要一条一条插入攒一批再插入使用连接池避免频繁创建连接读写分离主库写从库读考虑消息队列先写到消息队列再批量入库# 批量插入示例 def batch_insert_detections(detections, batch_size100): 批量插入检测结果 for i in range(0, len(detections), batch_size): batch detections[i:ibatch_size] # 执行批量插入 save_batch_to_db(batch) print(f已插入 {ilen(batch)}/{len(detections)} 条记录)6.3 数据一致性保证问题系统崩溃时有些检测结果没保存。解决方案使用事务确保数据完整性添加重试机制网络异常时自动重试记录日志方便排查问题定期备份防止数据丢失def save_with_retry(detection_data, max_retries3): 带重试的保存 for attempt in range(max_retries): try: # 开始事务 db.connection.start_transaction() # 保存数据 result_id db.save_detection_result(detection_data) # 提交事务 db.connection.commit() return result_id except Exception as e: # 回滚事务 db.connection.rollback() print(f第{attempt1}次尝试失败: {e}) if attempt max_retries - 1: # 最后一次尝试也失败记录到日志 log_failed_detection(detection_data) raise else: # 等待后重试 time.sleep(2 ** attempt) # 指数退避6.4 存储空间优化问题检测结果占用的存储空间增长太快。解决方案压缩存储对于不常用的字段可以用压缩格式存储归档旧数据定期将旧数据移到廉价存储清理无效数据删除置信度过低的检测结果使用合适的字段类型比如用SMALLINT代替INT-- 清理低置信度的检测结果比如低于0.3的 DELETE FROM detection_results WHERE confidence 0.3 AND detection_time DATE_SUB(NOW(), INTERVAL 7 DAY); -- 或者移到历史表 INSERT INTO detection_results_low_confidence SELECT * FROM detection_results WHERE confidence 0.3; DELETE FROM detection_results WHERE confidence 0.3;7. 监控与维护数据库运行起来后还需要定期监控和维护7.1 监控关键指标def check_database_health(self): 检查数据库健康状况 checks {} # 检查表大小 cursor self.db.connection.cursor() cursor.execute( SELECT table_name, ROUND(((data_length index_length) / 1024 / 1024), 2) as size_mb, table_rows FROM information_schema.tables WHERE table_schema DATABASE() AND table_name LIKE detection_% ORDER BY size_mb DESC ) checks[table_sizes] cursor.fetchall() # 检查索引使用情况 cursor.execute( SELECT index_name, non_unique, seq_in_index, column_name FROM information_schema.statistics WHERE table_schema DATABASE() AND table_name detection_results ORDER BY index_name, seq_in_index ) checks[indexes] cursor.fetchall() # 检查最近插入性能 cursor.execute( SELECT COUNT(*) as total_rows, MAX(detection_time) as latest_time, MIN(detection_time) as earliest_time, AVG(TIMESTAMPDIFF(SECOND, detection_time, NOW())) as avg_age_seconds FROM detection_results ) checks[data_stats] cursor.fetchone() cursor.close() return checks7.2 定期维护任务def perform_maintenance(self): 执行定期维护 maintenance_log [] try: # 1. 优化表 cursor self.db.connection.cursor() cursor.execute(OPTIMIZE TABLE detection_results) maintenance_log.append(表优化完成) # 2. 更新统计信息 cursor.execute(ANALYZE TABLE detection_results) maintenance_log.append(统计信息更新完成) # 3. 检查并修复表 cursor.execute(CHECK TABLE detection_results) result cursor.fetchone() if result[2] ! OK: cursor.execute(REPAIR TABLE detection_results) maintenance_log.append(表修复完成) # 4. 清理旧数据 cursor.execute( DELETE FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 90 DAY) AND confidence 0.5 ) deleted_rows cursor.rowcount maintenance_log.append(f清理了 {deleted_rows} 条旧记录) cursor.close() except Exception as e: maintenance_log.append(f维护过程中出错: {str(e)}) return maintenance_log8. 总结把YOLO12的检测结果存到MySQL里看起来简单实际上要考虑的细节还挺多的。从表结构设计、代码实现到查询优化、性能调优每个环节都关系到最终的使用体验。我建议你根据实际需求来调整比如数据量不大的话可以不用分区表查询不复杂的话索引也不用建太多。关键是找到适合自己项目的平衡点。实际用下来这套方案在大多数场景下都挺稳定的。数据存进去后查询起来确实方便多了做统计分析也简单。如果你刚开始做建议先按基础版本实现跑起来看看效果再根据实际情况调整优化。数据库这块水挺深的不同的使用场景、不同的数据量优化策略也不一样。最重要的是要结合实际业务需求来设计不要过度设计也不要设计不足。先跑起来再慢慢优化这样最稳妥。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
YOLO12模型在MySQL数据库中的结果存储与查询优化
YOLO12模型在MySQL数据库中的结果存储与查询优化如果你正在用YOLO12做目标检测每天处理成千上万张图片检测结果越来越多怎么存、怎么查就成了大问题。直接存成文件找起来麻烦分析起来更麻烦。用数据库怎么设计表结构才能既存得下又查得快今天咱们就来聊聊怎么把YOLO12的检测结果高效地存进MySQL还能让查询飞起来。我做了10多年AI项目踩过不少坑也总结了一些实用的经验分享给你。1. 为什么要把检测结果存进数据库你可能觉得检测结果不就是一些坐标和类别吗存成JSON或者CSV文件不就行了刚开始确实可以但项目做大了就会发现问题。我去年帮一个安防公司做项目他们用YOLO12做实时监控每天产生几十万条检测记录。刚开始他们用文件存储结果一个月后想查某个摄像头某段时间的检测记录得遍历几十个文件每次查询都要好几分钟。后来我们改成了数据库存储同样的查询现在只要几秒钟。数据库存储有几个明显的好处查询方便想查什么就查什么按时间、按类别、按摄像头随便组合统计分析可以轻松做各种统计比如“今天上午9点到10点3号摄像头检测到多少人”数据关联可以把检测结果和其他业务数据关联起来比如“检测到异常时自动触发报警”长期存储数据不会丢失方便回溯和分析2. 数据库表结构设计怎么存才合理设计表结构是个技术活设计得好查询快如闪电设计得不好查询慢如蜗牛。下面是我在实际项目中总结出来的设计思路。2.1 核心表设计先来看最核心的检测结果表。这个表要存每次检测的具体信息CREATE TABLE detection_results ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY COMMENT 主键ID, image_id VARCHAR(100) NOT NULL COMMENT 图片或视频帧的唯一标识, camera_id VARCHAR(50) COMMENT 摄像头编号, detection_time DATETIME NOT NULL COMMENT 检测时间, model_version VARCHAR(20) DEFAULT yolo12 COMMENT 模型版本, -- 检测对象信息 class_id INT NOT NULL COMMENT 类别ID, class_name VARCHAR(50) NOT NULL COMMENT 类别名称, confidence DECIMAL(5,4) NOT NULL COMMENT 置信度0.0000-1.0000, -- 边界框坐标归一化到0-1 x_center DECIMAL(6,5) NOT NULL COMMENT 中心点x坐标, y_center DECIMAL(6,5) NOT NULL COMMENT 中心点y坐标, width DECIMAL(6,5) NOT NULL COMMENT 边界框宽度, height DECIMAL(6,5) NOT NULL COMMENT 边界框高度, -- 原始像素坐标可选根据需求存储 pixel_x INT COMMENT 原始像素x坐标, pixel_y INT COMMENT 原始像素y坐标, pixel_width INT COMMENT 原始像素宽度, pixel_height INT COMMENT 原始像素高度, -- 其他信息 image_width INT COMMENT 图片宽度, image_height INT COMMENT 图片高度, processing_time_ms INT COMMENT 处理耗时毫秒, -- 索引字段 INDEX idx_detection_time (detection_time), INDEX idx_camera_time (camera_id, detection_time), INDEX idx_class_time (class_id, detection_time), INDEX idx_confidence (confidence) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENTYOLO12检测结果表;这个设计有几个关键点主键用自增ID查询快插入也快坐标存归一化值这样无论原始图片多大坐标都在0-1之间方便后续处理分开存类别ID和名称ID用于查询名称用于显示记录图片尺寸如果需要还原原始坐标有这个信息就行2.2 分类信息表如果类别比较多或者类别信息会变化最好单独建个表CREATE TABLE detection_classes ( class_id INT NOT NULL PRIMARY KEY COMMENT 类别ID, class_name VARCHAR(50) NOT NULL COMMENT 类别名称, display_name VARCHAR(50) COMMENT 显示名称, is_active TINYINT DEFAULT 1 COMMENT 是否启用, created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间, UNIQUE KEY uk_class_name (class_name), INDEX idx_is_active (is_active) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT检测类别表; -- 插入一些常见类别 INSERT INTO detection_classes (class_id, class_name, display_name) VALUES (0, person, 人), (1, bicycle, 自行车), (2, car, 汽车), (3, motorcycle, 摩托车), -- ... 其他YOLO12支持的类别 (56, chair, 椅子), (60, dining table, 餐桌);2.3 统计表可选如果经常需要做统计可以建个统计表每天或每小时更新一次CREATE TABLE detection_statistics ( stat_date DATE NOT NULL COMMENT 统计日期, camera_id VARCHAR(50) COMMENT 摄像头编号, class_id INT NOT NULL COMMENT 类别ID, detection_count INT DEFAULT 0 COMMENT 检测次数, avg_confidence DECIMAL(5,4) COMMENT 平均置信度, peak_hour TIME COMMENT 检测高峰时段, PRIMARY KEY (stat_date, camera_id, class_id), INDEX idx_date_camera (stat_date, camera_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT检测统计表;3. 从YOLO12到MySQL代码怎么写表设计好了接下来就是怎么把YOLO12的检测结果存进去。我用Python写了个简单的例子你可以参考一下。3.1 安装必要的库# 安装MySQL连接库 # pip install mysql-connector-python # 或者 # pip install pymysql # 安装YOLO12 # pip install ultralytics3.2 连接数据库的工具类import mysql.connector from mysql.connector import Error import json from datetime import datetime class DetectionDB: def __init__(self, hostlocalhost, databaseyolo_detection, userroot, passwordyour_password): self.host host self.database database self.user user self.password password self.connection None def connect(self): 连接到MySQL数据库 try: self.connection mysql.connector.connect( hostself.host, databaseself.database, userself.user, passwordself.password ) if self.connection.is_connected(): print(成功连接到MySQL数据库) return True except Error as e: print(f连接数据库时出错: {e}) return False def disconnect(self): 断开数据库连接 if self.connection and self.connection.is_connected(): self.connection.close() print(数据库连接已关闭) def save_detection_result(self, detection_data): 保存单条检测结果 if not self.connection: print(数据库未连接) return False try: cursor self.connection.cursor() # 构建插入SQL sql INSERT INTO detection_results ( image_id, camera_id, detection_time, model_version, class_id, class_name, confidence, x_center, y_center, width, height, pixel_x, pixel_y, pixel_width, pixel_height, image_width, image_height, processing_time_ms ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) # 执行插入 cursor.execute(sql, ( detection_data[image_id], detection_data.get(camera_id), detection_data[detection_time], detection_data.get(model_version, yolo12), detection_data[class_id], detection_data[class_name], float(detection_data[confidence]), float(detection_data[x_center]), float(detection_data[y_center]), float(detection_data[width]), float(detection_data[height]), detection_data.get(pixel_x), detection_data.get(pixel_y), detection_data.get(pixel_width), detection_data.get(pixel_height), detection_data.get(image_width), detection_data.get(image_height), detection_data.get(processing_time_ms) )) self.connection.commit() print(f检测结果已保存ID: {cursor.lastrowid}) cursor.close() return cursor.lastrowid except Error as e: print(f保存检测结果时出错: {e}) return False def batch_save_results(self, results_list): 批量保存检测结果 if not self.connection: print(数据库未连接) return False try: cursor self.connection.cursor() sql INSERT INTO detection_results ( image_id, camera_id, detection_time, model_version, class_id, class_name, confidence, x_center, y_center, width, height, pixel_x, pixel_y, pixel_width, pixel_height, image_width, image_height, processing_time_ms ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) # 准备数据 data_to_insert [] for detection in results_list: data_to_insert.append(( detection[image_id], detection.get(camera_id), detection[detection_time], detection.get(model_version, yolo12), detection[class_id], detection[class_name], float(detection[confidence]), float(detection[x_center]), float(detection[y_center]), float(detection[width]), float(detection[height]), detection.get(pixel_x), detection.get(pixel_y), detection.get(pixel_width), detection.get(pixel_height), detection.get(image_width), detection.get(image_height), detection.get(processing_time_ms) )) # 批量插入 cursor.executemany(sql, data_to_insert) self.connection.commit() print(f批量保存完成共插入 {cursor.rowcount} 条记录) cursor.close() return True except Error as e: print(f批量保存时出错: {e}) return False3.3 YOLO12检测结果处理from ultralytics import YOLO import cv2 from datetime import datetime import time class YOLO12Detector: def __init__(self, model_pathyolo12n.pt): 初始化YOLO12模型 print(正在加载YOLO12模型...) self.model YOLO(model_path) print(模型加载完成) # 连接到数据库 self.db DetectionDB() self.db.connect() def process_image(self, image_path, camera_idNone): 处理单张图片并保存结果到数据库 # 读取图片 image cv2.imread(image_path) if image is None: print(f无法读取图片: {image_path}) return image_height, image_width image.shape[:2] image_id fimg_{int(time.time())}_{hash(image_path) % 10000} # 使用YOLO12进行检测 start_time time.time() results self.model(image) processing_time int((time.time() - start_time) * 1000) # 毫秒 detection_time datetime.now() detection_results [] # 处理检测结果 for result in results: boxes result.boxes if boxes is not None: for box in boxes: # 获取检测信息 class_id int(box.cls[0]) class_name self.model.names[class_id] confidence float(box.conf[0]) # 获取边界框坐标归一化 x_center, y_center, width, height box.xywhn[0].tolist() # 计算原始像素坐标 pixel_x int(x_center * image_width - width * image_width / 2) pixel_y int(y_center * image_height - height * image_height / 2) pixel_width int(width * image_width) pixel_height int(height * image_height) # 构建检测数据 detection_data { image_id: image_id, camera_id: camera_id, detection_time: detection_time, model_version: yolo12, class_id: class_id, class_name: class_name, confidence: confidence, x_center: x_center, y_center: y_center, width: width, height: height, pixel_x: pixel_x, pixel_y: pixel_y, pixel_width: pixel_width, pixel_height: pixel_height, image_width: image_width, image_height: image_height, processing_time_ms: processing_time } detection_results.append(detection_data) # 保存到数据库 if detection_results: self.db.batch_save_results(detection_results) print(f图片 {image_path} 检测完成发现 {len(detection_results)} 个目标) else: print(f图片 {image_path} 未检测到目标) return detection_results def process_video_frame(self, frame, frame_id, camera_idNone): 处理视频帧 if frame is None: return [] image_height, image_width frame.shape[:2] image_id fframe_{camera_id}_{frame_id} if camera_id else fframe_{frame_id} # 使用YOLO12进行检测 start_time time.time() results self.model(frame) processing_time int((time.time() - start_time) * 1000) detection_time datetime.now() detection_results [] for result in results: boxes result.boxes if boxes is not None: for box in boxes: class_id int(box.cls[0]) class_name self.model.names[class_id] confidence float(box.conf[0]) x_center, y_center, width, height box.xywhn[0].tolist() detection_data { image_id: image_id, camera_id: camera_id, detection_time: detection_time, model_version: yolo12, class_id: class_id, class_name: class_name, confidence: confidence, x_center: x_center, y_center: y_center, width: width, height: height, image_width: image_width, image_height: image_height, processing_time_ms: processing_time } detection_results.append(detection_data) # 批量保存 if detection_results: self.db.batch_save_results(detection_results) return detection_results def close(self): 关闭资源 self.db.disconnect()3.4 使用示例# 使用示例 def main(): # 初始化检测器 detector YOLO12Detector(yolo12n.pt) try: # 处理单张图片 print(处理单张图片示例...) results detector.process_image(path/to/your/image.jpg, camera_idcam001) # 处理多张图片 image_paths [image1.jpg, image2.jpg, image3.jpg] for img_path in image_paths: detector.process_image(img_path, camera_idcam001) # 处理视频简化示例 print(\n处理视频示例...) import cv2 video_path path/to/your/video.mp4 cap cv2.VideoCapture(video_path) frame_id 0 while cap.isOpened(): ret, frame cap.read() if not ret: break # 每10帧处理一次根据实际需求调整 if frame_id % 10 0: detector.process_video_frame(frame, frame_id, camera_idcam002) frame_id 1 # 简单控制处理100帧后停止 if frame_id 100: break cap.release() finally: # 关闭资源 detector.close() if __name__ __main__: main()4. 查询优化让数据检索飞起来数据存进去了怎么查得快才是关键。下面分享几个实用的查询优化技巧。4.1 基础查询示例class DetectionQuery: def __init__(self, db_config): self.db DetectionDB(**db_config) self.db.connect() def get_recent_detections(self, limit100): 获取最近的检测记录 query SELECT dr.*, dc.display_name FROM detection_results dr LEFT JOIN detection_classes dc ON dr.class_id dc.class_id ORDER BY dr.detection_time DESC LIMIT %s cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (limit,)) results cursor.fetchall() cursor.close() return results def get_detections_by_class(self, class_name, start_timeNone, end_timeNone): 按类别查询检测记录 query SELECT * FROM detection_results WHERE class_name %s params [class_name] if start_time: query AND detection_time %s params.append(start_time) if end_time: query AND detection_time %s params.append(end_time) query ORDER BY detection_time DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, params) results cursor.fetchall() cursor.close() return results def get_detections_by_camera(self, camera_id, hours24): 查询指定摄像头最近N小时的检测记录 query SELECT class_name, COUNT(*) as count, AVG(confidence) as avg_confidence, DATE_FORMAT(detection_time, %%Y-%%m-%%d %%H:00:00) as hour_group FROM detection_results WHERE camera_id %s AND detection_time DATE_SUB(NOW(), INTERVAL %s HOUR) GROUP BY class_name, hour_group ORDER BY hour_group DESC, count DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (camera_id, hours)) results cursor.fetchall() cursor.close() return results def get_statistics_by_date(self, dateNone): 按日期统计检测结果 if date is None: date datetime.now().date() query SELECT class_name, COUNT(*) as total_count, AVG(confidence) as avg_confidence, MIN(confidence) as min_confidence, MAX(confidence) as max_confidence, COUNT(DISTINCT camera_id) as camera_count FROM detection_results WHERE DATE(detection_time) %s GROUP BY class_name ORDER BY total_count DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (date,)) results cursor.fetchall() cursor.close() return results4.2 高级查询技巧当数据量大的时候这些查询技巧能帮你节省很多时间def get_high_confidence_detections(self, min_confidence0.8, limit1000): 查询高置信度的检测记录使用索引 # 注意我们在表上创建了 idx_confidence 索引 query SELECT /* INDEX(detection_results idx_confidence) */ image_id, camera_id, detection_time, class_name, confidence, x_center, y_center FROM detection_results WHERE confidence %s ORDER BY confidence DESC, detection_time DESC LIMIT %s cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (min_confidence, limit)) return cursor.fetchall() def get_detections_in_area(self, x_min, x_max, y_min, y_max, start_timeNone, end_timeNone): 查询指定区域内的检测记录 # 这个查询可能比较慢需要根据实际情况优化 query SELECT * FROM detection_results WHERE x_center BETWEEN %s AND %s AND y_center BETWEEN %s AND %s AND confidence 0.5 params [x_min, x_max, y_min, y_max] if start_time: query AND detection_time %s params.append(start_time) if end_time: query AND detection_time %s params.append(end_time) query ORDER BY detection_time DESC LIMIT 1000 cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, params) return cursor.fetchall()4.3 数据库性能优化建议合理使用索引在经常查询的字段上创建索引比如 detection_time、camera_id、class_id复合索引要注意顺序比如 idx_camera_time (camera_id, detection_time)不要过度索引索引会影响插入速度分区表 如果数据量特别大比如每天百万条可以考虑按时间分区-- 按月分区 ALTER TABLE detection_results PARTITION BY RANGE (YEAR(detection_time) * 100 MONTH(detection_time)) ( PARTITION p202501 VALUES LESS THAN (202502), PARTITION p202502 VALUES LESS THAN (202503), PARTITION p202503 VALUES LESS THAN (202504), PARTITION p_future VALUES LESS THAN MAXVALUE );定期清理旧数据-- 删除3个月前的数据 DELETE FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 3 MONTH); -- 或者移到历史表 INSERT INTO detection_results_history SELECT * FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 6 MONTH); DELETE FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 6 MONTH);查询优化技巧避免 SELECT *只选择需要的字段使用 EXPLAIN 分析查询计划对于复杂查询考虑使用物化视图5. 实际应用场景5.1 安防监控系统在安防监控中我们可能需要实时查询某个区域的人员密度def get_person_density_by_area(self, camera_id, hours1): 查询指定摄像头最近1小时的人员密度变化 query SELECT DATE_FORMAT(detection_time, %%Y-%%m-%%d %%H:%%i:00) as minute_time, COUNT(*) as person_count, AVG(confidence) as avg_confidence FROM detection_results WHERE camera_id %s AND class_name person AND detection_time DATE_SUB(NOW(), INTERVAL %s HOUR) GROUP BY minute_time ORDER BY minute_time cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (camera_id, hours)) results cursor.fetchall() cursor.close() # 计算密度变化 density_data [] for row in results: density_data.append({ time: row[minute_time], count: row[person_count], density: row[person_count] / 100 # 假设每100像素一个人 }) return density_data5.2 零售客流分析在零售场景中可以分析顾客在店内的行为def analyze_customer_behavior(self, store_id, date): 分析顾客行为 query SELECT camera_id, class_name, HOUR(detection_time) as hour, COUNT(*) as detection_count, AVG(confidence) as avg_confidence, -- 计算平均停留位置 AVG(x_center) as avg_x, AVG(y_center) as avg_y FROM detection_results WHERE camera_id LIKE %s AND DATE(detection_time) %s AND class_name IN (person, shopping cart, bag) GROUP BY camera_id, class_name, HOUR(detection_time) ORDER BY camera_id, hour, class_name cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (f{store_id}%, date)) results cursor.fetchall() cursor.close() # 按小时分析 hourly_analysis {} for row in results: hour row[hour] if hour not in hourly_analysis: hourly_analysis[hour] {person: 0, other: 0} if row[class_name] person: hourly_analysis[hour][person] row[detection_count] else: hourly_analysis[hour][other] row[detection_count] return { detail: results, hourly_summary: hourly_analysis }5.3 工业质检系统在工业质检中需要快速查询缺陷产品的检测记录def get_defect_statistics(self, production_line, start_date, end_date): 获取缺陷产品统计 query SELECT class_name as defect_type, COUNT(*) as defect_count, DATE(detection_time) as defect_date, camera_id as station, -- 计算缺陷率 ROUND(COUNT(*) * 100.0 / ( SELECT COUNT(*) FROM detection_results WHERE camera_id dr.camera_id AND DATE(detection_time) DATE(dr.detection_time) AND class_name ! ok ), 2) as defect_rate FROM detection_results dr WHERE camera_id LIKE %s AND detection_time BETWEEN %s AND %s AND class_name ! ok -- 假设ok表示合格品 AND confidence 0.7 GROUP BY defect_type, defect_date, station ORDER BY defect_date DESC, defect_count DESC cursor self.db.connection.cursor(dictionaryTrue) cursor.execute(query, (fline{production_line}_%, start_date, end_date)) results cursor.fetchall() cursor.close() return results6. 常见问题与解决方案在实际使用中你可能会遇到这些问题6.1 数据量太大查询慢怎么办问题检测结果表有上千万条记录查询越来越慢。解决方案分区表按时间分区比如每个月一个分区归档历史数据把3个月前的数据移到历史表添加合适索引在经常查询的字段上加索引使用查询缓存对于不经常变动的统计查询可以用缓存-- 示例创建月度分区 ALTER TABLE detection_results PARTITION BY RANGE (TO_DAYS(detection_time)) ( PARTITION p202501 VALUES LESS THAN (TO_DAYS(2025-02-01)), PARTITION p202502 VALUES LESS THAN (TO_DAYS(2025-03-01)), PARTITION p202503 VALUES LESS THAN (TO_DAYS(2025-04-01)) );6.2 并发写入性能问题问题多个摄像头同时写入数据库压力大。解决方案批量插入不要一条一条插入攒一批再插入使用连接池避免频繁创建连接读写分离主库写从库读考虑消息队列先写到消息队列再批量入库# 批量插入示例 def batch_insert_detections(detections, batch_size100): 批量插入检测结果 for i in range(0, len(detections), batch_size): batch detections[i:ibatch_size] # 执行批量插入 save_batch_to_db(batch) print(f已插入 {ilen(batch)}/{len(detections)} 条记录)6.3 数据一致性保证问题系统崩溃时有些检测结果没保存。解决方案使用事务确保数据完整性添加重试机制网络异常时自动重试记录日志方便排查问题定期备份防止数据丢失def save_with_retry(detection_data, max_retries3): 带重试的保存 for attempt in range(max_retries): try: # 开始事务 db.connection.start_transaction() # 保存数据 result_id db.save_detection_result(detection_data) # 提交事务 db.connection.commit() return result_id except Exception as e: # 回滚事务 db.connection.rollback() print(f第{attempt1}次尝试失败: {e}) if attempt max_retries - 1: # 最后一次尝试也失败记录到日志 log_failed_detection(detection_data) raise else: # 等待后重试 time.sleep(2 ** attempt) # 指数退避6.4 存储空间优化问题检测结果占用的存储空间增长太快。解决方案压缩存储对于不常用的字段可以用压缩格式存储归档旧数据定期将旧数据移到廉价存储清理无效数据删除置信度过低的检测结果使用合适的字段类型比如用SMALLINT代替INT-- 清理低置信度的检测结果比如低于0.3的 DELETE FROM detection_results WHERE confidence 0.3 AND detection_time DATE_SUB(NOW(), INTERVAL 7 DAY); -- 或者移到历史表 INSERT INTO detection_results_low_confidence SELECT * FROM detection_results WHERE confidence 0.3; DELETE FROM detection_results WHERE confidence 0.3;7. 监控与维护数据库运行起来后还需要定期监控和维护7.1 监控关键指标def check_database_health(self): 检查数据库健康状况 checks {} # 检查表大小 cursor self.db.connection.cursor() cursor.execute( SELECT table_name, ROUND(((data_length index_length) / 1024 / 1024), 2) as size_mb, table_rows FROM information_schema.tables WHERE table_schema DATABASE() AND table_name LIKE detection_% ORDER BY size_mb DESC ) checks[table_sizes] cursor.fetchall() # 检查索引使用情况 cursor.execute( SELECT index_name, non_unique, seq_in_index, column_name FROM information_schema.statistics WHERE table_schema DATABASE() AND table_name detection_results ORDER BY index_name, seq_in_index ) checks[indexes] cursor.fetchall() # 检查最近插入性能 cursor.execute( SELECT COUNT(*) as total_rows, MAX(detection_time) as latest_time, MIN(detection_time) as earliest_time, AVG(TIMESTAMPDIFF(SECOND, detection_time, NOW())) as avg_age_seconds FROM detection_results ) checks[data_stats] cursor.fetchone() cursor.close() return checks7.2 定期维护任务def perform_maintenance(self): 执行定期维护 maintenance_log [] try: # 1. 优化表 cursor self.db.connection.cursor() cursor.execute(OPTIMIZE TABLE detection_results) maintenance_log.append(表优化完成) # 2. 更新统计信息 cursor.execute(ANALYZE TABLE detection_results) maintenance_log.append(统计信息更新完成) # 3. 检查并修复表 cursor.execute(CHECK TABLE detection_results) result cursor.fetchone() if result[2] ! OK: cursor.execute(REPAIR TABLE detection_results) maintenance_log.append(表修复完成) # 4. 清理旧数据 cursor.execute( DELETE FROM detection_results WHERE detection_time DATE_SUB(NOW(), INTERVAL 90 DAY) AND confidence 0.5 ) deleted_rows cursor.rowcount maintenance_log.append(f清理了 {deleted_rows} 条旧记录) cursor.close() except Exception as e: maintenance_log.append(f维护过程中出错: {str(e)}) return maintenance_log8. 总结把YOLO12的检测结果存到MySQL里看起来简单实际上要考虑的细节还挺多的。从表结构设计、代码实现到查询优化、性能调优每个环节都关系到最终的使用体验。我建议你根据实际需求来调整比如数据量不大的话可以不用分区表查询不复杂的话索引也不用建太多。关键是找到适合自己项目的平衡点。实际用下来这套方案在大多数场景下都挺稳定的。数据存进去后查询起来确实方便多了做统计分析也简单。如果你刚开始做建议先按基础版本实现跑起来看看效果再根据实际情况调整优化。数据库这块水挺深的不同的使用场景、不同的数据量优化策略也不一样。最重要的是要结合实际业务需求来设计不要过度设计也不要设计不足。先跑起来再慢慢优化这样最稳妥。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
