diff --git a/app.py b/app.py index f121af8..247b982 100644 --- a/app.py +++ b/app.py @@ -12,7 +12,7 @@ import pytz from lib.db import execute_query from lib.report_generator import generate_daily_report from lib.api import register_blueprints -from lib.db_init import init_database_tables +from lib.db_init import init_database_tables, optimize_sum_data_indexes from lib.logger import log_info, log_error, log_warning from lib.api.download_api import cleanup_old_temp_zips from lib.api.auth_api import get_current_user @@ -33,6 +33,13 @@ register_blueprints(app) log_info('正在初始化数据库表...', 'app') if init_database_tables(): log_info('数据库表初始化成功', 'app') + + # 优化求和数据表索引 + log_info('正在优化数据库索引...', 'app') + if optimize_sum_data_indexes(): + log_info('数据库索引优化成功', 'app') + else: + log_warning('数据库索引优化失败，但不影响正常使用', 'app') else: log_error('数据库表初始化失败，请检查数据库连接', 'app') @@ -145,12 +152,40 @@ def scheduled_job(): log_error(f'定时任务执行失败：{e}', 'scheduler') -# 启动定时任务 +# ==================== 启动定时任务 ==================== scheduler = BackgroundScheduler(timezone='Asia/Shanghai') scheduler.add_job(scheduled_job, 'cron', hour=8, minute=30) scheduler.add_job(cleanup_old_temp_zips, 'interval', hours=1) + +# 添加 Doris 表统计信息定时优化任务（每天凌晨3点执行） +def optimize_doris_tables(): + """定期优化 Doris 表统计信息""" + try: + log_info('开始执行 Doris 表统计信息优化...', 'scheduler') + from lib.doris_optimize import optimize_doris_table + + tables = [ + 't_daily_report_sum_data', + 't_daily_report_config', + 't_daily_report_history', + ] + + for table in tables: + try: + optimize_doris_table(table) + log_info(f'表 {table} 统计信息优化完成', 'scheduler') + except Exception as e: + log_warning(f'表 {table} 统计信息优化失败: {e}', 'scheduler') + + log_info('Doris 表统计信息优化完成', 'scheduler') + except Exception as e: + log_error(f'Doris 表统计信息优化失败: {e}', 'scheduler') + + +scheduler.add_job(optimize_doris_tables, 'cron', hour=3, minute=0) + scheduler.start() -log_info('定时任务已启动：每天 08:30 执行日报生成，每小时清理临时文件', 'app') +log_info('定时任务已启动：每天 08:30 执行日报生成，每小时清理临时文件，每天 03:00 优化 Doris 统计信息', 'app') # ==================== 启动应用 ==================== diff --git a/lib/api/global_config_api.py b/lib/api/global_config_api.py index b280a29..ff0bbd2 100644 --- a/lib/api/global_config_api.py +++ b/lib/api/global_config_api.py @@ -1,5 +1,6 @@ from flask import Blueprint, jsonify, request from lib.db import execute_query, execute_update +from lib.doris_optimize import get_pool, optimize_doris_table, get_doris_table_stats import json global_config_api = Blueprint('global_config_api', __name__) @@ -33,4 +34,66 @@ def save_sum_data_fields(): return jsonify({'success': True, 'message': '配置保存成功'}) except Exception as e: - return jsonify({'success': False, 'message': str(e)}), 500 \ No newline at end of file + return jsonify({'success': False, 'message': str(e)}), 500 + + +# ==================== Doris 优化接口 ==================== + +@global_config_api.route('/api/doris/status', methods=['GET']) +def get_doris_status(): + """获取 Doris 连接池状态""" + try: + pool = get_pool() + stats = pool.get_stats() + return jsonify({'success': True, 'data': stats}) + except Exception as e: + return jsonify({'success': False, 'message': str(e)}), 500 + + +@global_config_api.route('/api/doris/optimize-table', methods=['POST']) +def api_optimize_table(): + """优化 Doris 表统计信息""" + try: + data = request.get_json() + table_name = data.get('table_name', '') + + if not table_name: + return jsonify({'success': False, 'message': '请提供表名'}), 400 + + result = optimize_doris_table(table_name) + if result: + return jsonify({'success': True, 'message': f'表 {table_name} 优化成功'}) + else: + return jsonify({'success': False, 'message': f'表 {table_name} 优化失败'}) + except Exception as e: + return jsonify({'success': False, 'message': str(e)}), 500 + + +@global_config_api.route('/api/doris/table-stats', methods=['GET']) +def api_get_table_stats(): + """获取 Doris 表统计信息""" + try: + table_name = request.args.get('table_name', 't_daily_report_sum_data') + stats = get_doris_table_stats(table_name) + return jsonify({'success': True, 'data': stats}) + except Exception as e: + return jsonify({'success': False, 'message': str(e)}), 500 + + +@global_config_api.route('/api/doris/health-check', methods=['GET']) +def doris_health_check(): + """Doris 健康检查""" + try: + # 简单查询测试连接 + result = execute_query('SELECT 1 as test') + return jsonify({ + 'success': True, + 'status': 'healthy', + 'test_query': result + }) + except Exception as e: + return jsonify({ + 'success': False, + 'status': 'unhealthy', + 'message': str(e) + }), 500 \ No newline at end of file diff --git a/lib/api/sum_data_api.py b/lib/api/sum_data_api.py index 9d014e1..3bf5adf 100644 --- a/lib/api/sum_data_api.py +++ b/lib/api/sum_data_api.py @@ -51,7 +51,11 @@ def list_sum_data(): config_name = request.args.get('config_name') sum_field_key = request.args.get('sum_field_key') data_source = request.args.get('data_source') - log_info(f'[求和数据API] 查询求和数据列表，page={page}, page_size={page_size}', 'api') + + # 当 page_size 较大时（>=1000），跳过 COUNT 查询以优化性能 + skip_count = page_size >= 1000 + + log_info(f'[求和数据API] 查询求和数据列表，page={page}, page_size={page_size}, skip_count={skip_count}', 'api') result = get_sum_data_list( page=page, @@ -61,11 +65,12 @@ def list_sum_data(): config_id=config_id, config_name=config_name, sum_field_key=sum_field_key, - data_source=data_source + data_source=data_source, + skip_count=skip_count ) if result['success']: - log_info(f'[求和数据API] 查询成功，共{result["data"]["total"]}条记录', 'api') + log_info(f'[求和数据API] 查询成功，共{len(result["data"]["list"])}条记录', 'api') return jsonify(result) else: return jsonify(result), 400 diff --git a/lib/db.py b/lib/db.py index 9b91e11..8cb9d76 100644 --- a/lib/db.py +++ b/lib/db.py @@ -1,10 +1,14 @@ """ 数据库连接配置 +针对 Apache Doris 优化的数据库操作模块 """ import os +import time import pymysql from contextlib import contextmanager from lib.logger import log_info, log_error, log_warning +from lib.doris_optimize import get_pool, get_doris_connection, execute_doris_query, execute_doris_update, execute_doris_batch_insert + def get_db_config(): """从环境变量或默认值获取数据库配置""" @@ -17,7 +21,7 @@ def get_db_config(): 'charset': 'utf8mb4', 'cursorclass': pymysql.cursors.DictCursor, 'autocommit': True, - 'connect_timeout': 30, + 'connect_timeout': 10, # 优化：缩短连接超时 'read_timeout': 120, 'write_timeout': 120 } @@ -27,86 +31,79 @@ DB_CONFIG = get_db_config() @contextmanager def get_connection(): - """获取数据库连接（上下文管理器）""" + """获取数据库连接（上下文管理器）- 优先使用连接池""" try: - conn = pymysql.connect(**DB_CONFIG) - yield conn - except pymysql.Error as e: - log_error(f'数据库连接失败: {e}', 'db', exc_info=True) - raise - finally: + # 使用连接池获取连接 + with get_doris_connection() as conn: + yield conn + except Exception: + # 连接池失败时回退到直接连接 + log_warning('[数据库] 连接池获取失败，回退到直接连接', 'db') try: - conn.close() - except: - pass + conn = pymysql.connect(**DB_CONFIG) + yield conn + except pymysql.Error as e: + log_error(f'数据库连接失败: {e}', 'db', exc_info=True) + raise + finally: + try: + conn.close() + except: + pass def execute_query(sql, params=None, retry=2): - """执行查询并返回结果""" - for attempt in range(retry + 1): - try: - with get_connection() as conn: - with conn.cursor() as cursor: - cursor.execute(sql, params) - result = cursor.fetchall() - if attempt > 0: - log_info(f'[数据库] 查询重试成功，第{attempt+1}次尝试', 'db') - return result - except pymysql.err.OperationalError as e: - if attempt < retry and (e.args[0] == 2013 or e.args[0] == 2006): - log_warning(f'[数据库] 查询连接断开，正在重试（第{attempt+1}次）: {e}', 'db') - import time - time.sleep(1) - continue - log_error(f'查询失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True) - raise - except Exception as e: - log_error(f'查询失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True) - raise + """执行查询并返回结果 - 使用 Doris 优化版本""" + return execute_doris_query(sql, params, retry) def execute_update(sql, params=None, retry=2): - """执行更新操作""" - for attempt in range(retry + 1): - try: - with get_connection() as conn: - with conn.cursor() as cursor: - affected = cursor.execute(sql, params) - if attempt > 0: - log_info(f'[数据库] 更新重试成功，第{attempt+1}次尝试', 'db') - return cursor.rowcount - except pymysql.err.OperationalError as e: - if attempt < retry and (e.args[0] == 2013 or e.args[0] == 2006): - log_warning(f'[数据库] 更新连接断开，正在重试（第{attempt+1}次）: {e}', 'db') - import time - time.sleep(1) - continue - log_error(f'更新失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True) - raise - except Exception as e: - log_error(f'更新失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True) - raise + """执行更新操作 - 使用 Doris 优化版本""" + return execute_doris_update(sql, params, retry) def execute_insert(sql, params=None, retry=2): """执行插入操作并返回插入ID""" for attempt in range(retry + 1): try: - with get_connection() as conn: + start_time = time.time() + + with get_doris_connection() as conn: with conn.cursor() as cursor: cursor.execute(sql, params) last_id = cursor.lastrowid - if attempt > 0: - log_info(f'[数据库] 插入重试成功，第{attempt+1}次尝试', 'db') - return last_id + + elapsed = (time.time() - start_time) * 1000 + + if elapsed > 100: + log_info(f"[Doris慢插入] 耗时: {elapsed:.1f}ms, SQL: {sql[:200]}", 'doris') + + if attempt > 0: + log_info(f'[数据库] 插入重试成功，第{attempt+1}次尝试', 'db') + + return last_id except pymysql.err.OperationalError as e: if attempt < retry and (e.args[0] == 2013 or e.args[0] == 2006): log_warning(f'[数据库] 插入连接断开，正在重试（第{attempt+1}次）: {e}', 'db') - import time - time.sleep(1) + time.sleep(0.5 * (attempt + 1)) continue log_error(f'插入失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True) raise except Exception as e: log_error(f'插入失败: {e}\nSQL: {sql}\nParams: {params}', 'db', exc_info=True) - raise \ No newline at end of file + raise + + +def execute_batch_insert(sql, params_list, batch_size=100): + """ + 批量插入优化（针对 Doris 特性优化） + + Args: + sql: SQL 语句 + params_list: 参数列表 + batch_size: 每批大小（默认100） + + Returns: + int: 插入的总行数 + """ + return execute_doris_batch_insert(sql, params_list, batch_size) \ No newline at end of file diff --git a/lib/db_init.py b/lib/db_init.py index 1335842..a15f4f9 100644 --- a/lib/db_init.py +++ b/lib/db_init.py @@ -275,3 +275,133 @@ def init_database_tables(): except Exception as e: log_error(f"[数据库] 初始化失败: {e}", 'db_init', exc_info=True) return False + + +def optimize_sum_data_indexes(): + """优化求和数据表的索引，提升查询性能""" + try: + log_info("[数据库] 开始优化 t_daily_report_sum_data 表索引", 'db_init') + + # 检查倒排索引是否存在 + def _index_exists(index_name): + try: + result = execute_query("SHOW INDEX FROM t_daily_report_sum_data") + for row in (result or []): + key_name = row.get('Key_name', '') or row.get('Index_name', '') + if key_name == index_name: + return True + except Exception: + pass + return False + + # 添加倒排索引以加速查询 + # report_date 索引 - 日期范围查询 + try: + if not _index_exists('idx_report_date'): + execute_update(""" + ALTER TABLE t_daily_report_sum_data + ADD INDEX idx_report_date (report_date) + USING INVERTED PROPERTIES("parser" = "none") + COMMENT '报表日期索引' + """) + log_info("[数据库] 添加 report_date 倒排索引", 'db_init') + else: + log_info("[数据库] report_date 倒排索引已存在", 'db_init') + except Exception as e: + log_warning(f"[数据库] 添加 report_date 索引失败: {e}", 'db_init') + + # config_id 索引 - 配置查询 + try: + if not _index_exists('idx_config_id'): + execute_update(""" + ALTER TABLE t_daily_report_sum_data + ADD INDEX idx_config_id (config_id) + USING INVERTED PROPERTIES("parser" = "none") + COMMENT '配置ID索引' + """) + log_info("[数据库] 添加 config_id 倒排索引", 'db_init') + else: + log_info("[数据库] config_id 倒排索引已存在", 'db_init') + except Exception as e: + log_warning(f"[数据库] 添加 config_id 索引失败: {e}", 'db_init') + + # data_source 索引 - 数据来源查询 + try: + if not _index_exists('idx_data_source'): + execute_update(""" + ALTER TABLE t_daily_report_sum_data + ADD INDEX idx_data_source (data_source) + USING INVERTED PROPERTIES("parser" = "none") + COMMENT '数据来源索引' + """) + log_info("[数据库] 添加 data_source 倒排索引", 'db_init') + else: + log_info("[数据库] data_source 倒排索引已存在", 'db_init') + except Exception as e: + log_warning(f"[数据库] 添加 data_source 索引失败: {e}", 'db_init') + + # sum_field_key 索引 - 字段查询 + try: + if not _index_exists('idx_sum_field_key'): + execute_update(""" + ALTER TABLE t_daily_report_sum_data + ADD INDEX idx_sum_field_key (sum_field_key) + USING INVERTED PROPERTIES("parser" = "none") + COMMENT '求和字段索引' + """) + log_info("[数据库] 添加 sum_field_key 倒排索引", 'db_init') + else: + log_info("[数据库] sum_field_key 倒排索引已存在", 'db_init') + except Exception as e: + log_warning(f"[数据库] 添加 sum_field_key 索引失败: {e}", 'db_init') + + # config_name 索引 - 配置名称模糊搜索 + try: + if not _index_exists('idx_config_name'): + execute_update(""" + ALTER TABLE t_daily_report_sum_data + ADD INDEX idx_config_name (config_name) + USING INVERTED PROPERTIES("parser" = "none") + COMMENT '配置名称索引' + """) + log_info("[数据库] 添加 config_name 倒排索引", 'db_init') + else: + log_info("[数据库] config_name 倒排索引已存在", 'db_init') + except Exception as e: + log_warning(f"[数据库] 添加 config_name 索引失败: {e}", 'db_init') + + # split_type 索引 - 拆分类型查询 + try: + if not _index_exists('idx_split_type'): + execute_update(""" + ALTER TABLE t_daily_report_sum_data + ADD INDEX idx_split_type (split_type) + USING INVERTED PROPERTIES("parser" = "none") + COMMENT '拆分类型索引' + """) + log_info("[数据库] 添加 split_type 倒排索引", 'db_init') + else: + log_info("[数据库] split_type 倒排索引已存在", 'db_init') + except Exception as e: + log_warning(f"[数据库] 添加 split_type 索引失败: {e}", 'db_init') + + # split_value 索引 - 拆分值查询 + try: + if not _index_exists('idx_split_value'): + execute_update(""" + ALTER TABLE t_daily_report_sum_data + ADD INDEX idx_split_value (split_value) + USING INVERTED PROPERTIES("parser" = "none") + COMMENT '拆分值索引' + """) + log_info("[数据库] 添加 split_value 倒排索引", 'db_init') + else: + log_info("[数据库] split_value 倒排索引已存在", 'db_init') + except Exception as e: + log_warning(f"[数据库] 添加 split_value 索引失败: {e}", 'db_init') + + log_info("[数据库] 求和数据表索引优化完成", 'db_init') + return True + except Exception as e: + log_error(f"[数据库] 优化索引失败: {e}", 'db_init', exc_info=True) + return False diff --git a/lib/doris_optimize.py b/lib/doris_optimize.py new file mode 100644 index 0000000..eb5ae45 --- /dev/null +++ b/lib/doris_optimize.py @@ -0,0 +1,380 @@ +""" +Apache Doris 数据库优化模块 +针对 Doris 特性的优化策略： +1. 连接池配置优化 +2. 批量操作优化 +3. 查询性能监控 +4. 表结构优化建议 +""" +import os +import time +import pymysql +from contextlib import contextmanager +from collections import deque +from threading import Lock +from lib.logger import log_info, log_error, log_warning + + +class DorisConnectionPool: + """Doris 数据库连接池""" + + _instance = None + _lock = Lock() + + def __new__(cls, *args, **kwargs): + if cls._instance is None: + with cls._lock: + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self, host=None, port=None, user=None, password=None, + database=None, pool_size=10, **kwargs): + if self._initialized: + return + + self._host = host or os.environ.get('DB_HOST', 'haoslm2.xicp.net') + self._port = port or int(os.environ.get('DB_PORT', 10216)) + self._user = user or os.environ.get('DB_USER', 'root') + self._password = password or os.environ.get('DB_PASSWORD', 'DsideaL147258369') + self._database = database or os.environ.get('DB_NAME', 'yltcharge') + self._pool_size = pool_size + + # 连接池 + self._pool = deque(maxlen=pool_size) + self._pool_lock = Lock() + self._total_connections = 0 + self._initialized = True + + # 预热连接池 + self._preheat_pool() + log_info(f"[Doris连接池] 初始化完成，目标连接数: {pool_size}", 'doris') + + def _preheat_pool(self): + """预热连接池""" + try: + for i in range(min(3, self._pool_size)): + try: + conn = self._create_connection() + with self._pool_lock: + self._pool.append(conn) + self._total_connections += 1 + log_info(f"[Doris连接池] 预热连接 {i+1}/3 成功", 'doris') + except Exception as e: + log_warning(f"[Doris连接池] 预热连接失败: {e}", 'doris') + except Exception as e: + log_warning(f"[Doris连接池] 预热异常: {e}", 'doris') + + def _create_connection(self): + """创建新连接""" + conn = pymysql.connect( + host=self._host, + port=self._port, + user=self._user, + password=self._password, + database=self._database, + charset='utf8mb4', + cursorclass=pymysql.cursors.DictCursor, + autocommit=True, + connect_timeout=10, # 缩短连接超时 + read_timeout=120, + write_timeout=120, + # Doris 特有优化 + init_command='SET SESSION query_timeout = 300000', # 5分钟超时 + ) + return conn + + def get_connection(self, timeout=5): + """获取连接（带超时）""" + start_time = time.time() + + # 先从池中获取 + with self._pool_lock: + if self._pool: + conn = self._pool.pop() + # 检查连接是否有效 + try: + conn.ping(reconnect=False) + return conn + except Exception: + # 连接失效，创建新连接 + self._total_connections -= 1 + try: + conn.close() + except Exception: + pass + + # 池没有可用连接，等待或创建新连接 + elapsed = time.time() - start_time + if elapsed < timeout: + # 尝试创建新连接（如果总数未超过限制） + if self._total_connections < self._pool_size: + try: + conn = self._create_connection() + with self._pool_lock: + self._total_connections += 1 + return conn + except Exception as e: + log_warning(f"[Doris连接池] 创建新连接失败: {e}", 'doris') + + # 超时或达到上限，强制创建新连接 + log_warning(f"[Doris连接池] 强制创建新连接（池已满或超时）", 'doris') + return self._create_connection() + + def return_connection(self, conn): + """归还连接到池""" + if conn is None: + return + try: + conn.ping(reconnect=False) + with self._pool_lock: + if len(self._pool) < self._pool_size: + self._pool.append(conn) + return + except Exception: + pass + + # 连接失效或池已满，关闭连接 + try: + conn.close() + except Exception: + pass + with self._pool_lock: + self._total_connections = max(0, self._total_connections - 1) + + def get_stats(self): + """获取连接池状态""" + with self._pool_lock: + return { + 'pool_size': len(self._pool), + 'total_connections': self._total_connections, + 'max_pool_size': self._pool_size + } + + +# 全局连接池实例 +_pool = None + + +def get_pool(): + """获取全局连接池""" + global _pool + if _pool is None: + _pool = DorisConnectionPool(pool_size=10) + return _pool + + +@contextmanager +def get_doris_connection(): + """获取 Doris 连接的上下文管理器""" + pool = get_pool() + conn = pool.get_connection(timeout=3) + try: + yield conn + except pymysql.Error as e: + log_error(f"[Doris] 数据库操作失败: {e}", 'doris') + # 连接可能已失效，不归还到池 + try: + conn.close() + except Exception: + pass + with pool._pool_lock: + pool._total_connections = max(0, pool._total_connections - 1) + raise + else: + pool.return_connection(conn) + + +def execute_doris_query(sql, params=None, retry=2): + """ + 执行 Doris 查询（带重试和性能监控） + + Args: + sql: SQL 语句 + params: 参数 + retry: 重试次数 + + Returns: + list: 查询结果 + """ + for attempt in range(retry + 1): + try: + start_time = time.time() + + with get_doris_connection() as conn: + with conn.cursor() as cursor: + cursor.execute(sql, params) + result = cursor.fetchall() + + elapsed = (time.time() - start_time) * 1000 # 毫秒 + + # 慢查询监控（超过100ms记录） + if elapsed > 100: + log_info(f"[Doris慢查询] 耗时: {elapsed:.1f}ms, SQL: {sql[:200]}, 行数: {len(result) if result else 0}", 'doris') + + if attempt > 0: + log_info(f'[Doris] 查询重试成功，第{attempt+1}次尝试', 'doris') + + return result + + except pymysql.err.OperationalError as e: + if attempt < retry and (e.args[0] == 2013 or e.args[0] == 2006): + log_warning(f'[Doris] 查询连接断开，正在重试（第{attempt+1}次）: {e}', 'doris') + time.sleep(0.5 * (attempt + 1)) + continue + log_error(f'[Doris] 查询失败: {e}\nSQL: {sql}\nParams: {params}', 'doris') + raise + except Exception as e: + log_error(f'[Doris] 查询失败: {e}\nSQL: {sql}\nParams: {params}', 'doris') + raise + + +def execute_doris_batch_insert(sql, params_list, batch_size=100, retry=2): + """ + Doris 批量插入优化 + + Args: + sql: SQL 语句（带 %s 占位符） + params_list: 参数列表 + batch_size: 每批大小 + retry: 重试次数 + + Returns: + int: 插入的总行数 + """ + if not params_list: + return 0 + + total_inserted = 0 + start_time = time.time() + + for i in range(0, len(params_list), batch_size): + batch = params_list[i:i + batch_size] + batch_sql = sql.rstrip() + + # 对于 Doris，可以使用 INSERT INTO ... VALUES (...), (...), (...) 格式 + # 但 pymysql 的 executemany 已经处理好了 + + for attempt in range(retry + 1): + try: + with get_doris_connection() as conn: + with conn.cursor() as cursor: + cursor.executemany(batch_sql, batch) + total_inserted += len(batch) + + if attempt > 0: + log_info(f'[Doris] 批量插入重试成功，第{attempt+1}次尝试', 'doris') + break + + except pymysql.err.OperationalError as e: + if attempt < retry and (e.args[0] == 2013 or e.args[0] == 2006): + log_warning(f'[Doris] 批量插入连接断开，正在重试（第{attempt+1}次）: {e}', 'doris') + time.sleep(0.5 * (attempt + 1)) + continue + log_error(f'[Doris] 批量插入失败: {e}\nSQL: {batch_sql}', 'doris') + raise + except Exception as e: + log_error(f'[Doris] 批量插入失败: {e}\nSQL: {batch_sql}', 'doris') + raise + + elapsed = (time.time() - start_time) * 1000 + log_info(f'[Doris] 批量插入完成，共 {total_inserted} 行，耗时 {elapsed:.1f}ms', 'doris') + + return total_inserted + + +def execute_doris_update(sql, params=None, retry=2): + """ + 执行 Doris 更新操作（带重试） + + Args: + sql: SQL 语句 + params: 参数 + retry: 重试次数 + + Returns: + int: 影响的行数 + """ + for attempt in range(retry + 1): + try: + start_time = time.time() + + with get_doris_connection() as conn: + with conn.cursor() as cursor: + affected = cursor.execute(sql, params) + + elapsed = (time.time() - start_time) * 1000 + + if elapsed > 100: + log_info(f"[Doris慢更新] 耗时: {elapsed:.1f}ms, SQL: {sql[:200]}", 'doris') + + if attempt > 0: + log_info(f'[Doris] 更新重试成功，第{attempt+1}次尝试', 'doris') + + return cursor.rowcount if hasattr(cursor, 'rowcount') else affected + + except pymysql.err.OperationalError as e: + if attempt < retry and (e.args[0] == 2013 or e.args[0] == 2006): + log_warning(f'[Doris] 更新连接断开，正在重试（第{attempt+1}次）: {e}', 'doris') + time.sleep(0.5 * (attempt + 1)) + continue + log_error(f'[Doris] 更新失败: {e}\nSQL: {sql}\nParams: {params}', 'doris') + raise + except Exception as e: + log_error(f'[Doris] 更新失败: {e}\nSQL: {sql}\nParams: {params}', 'doris') + raise + + +def optimize_doris_table(table_name): + """ + 优化 Doris 表的统计信息 + + Args: + table_name: 表名 + """ + try: + sql = f"ANALYZE TABLE {table_name} UPDATE HISTOGRAM" + execute_doris_update(sql) + log_info(f"[Doris] 表 {table_name} 统计信息已更新", 'doris') + return True + except Exception as e: + log_warning(f"[Doris] 表 {table_name} 统计信息更新失败: {e}", 'doris') + # 尝试使用 COMPUTE STATISTICS + try: + sql = f"ANALYZE TABLE {table_name} COMPUTE STATISTICS" + execute_doris_update(sql) + log_info(f"[Doris] 表 {table_name} 统计信息已计算", 'doris') + return True + except Exception as e2: + log_warning(f"[Doris] 表 {table_name} 统计信息计算也失败: {e2}", 'doris') + return False + + +def get_doris_table_stats(table_name): + """ + 获取 Doris 表的统计信息 + + Args: + table_name: 表名 + + Returns: + dict: 统计信息 + """ + try: + sql = f"SHOW TABLE STATUS LIKE '{table_name}'" + result = execute_doris_query(sql) + if result: + return result[0] + return {} + except Exception as e: + log_warning(f"[Doris] 获取表 {table_name} 统计信息失败: {e}", 'doris') + return {} + + +def print_pool_stats(): + """打印连接池状态""" + pool = get_pool() + stats = pool.get_stats() + log_info(f"[Doris连接池] 状态: {stats}", 'doris') + return stats diff --git a/lib/sum_data_collector.py b/lib/sum_data_collector.py index 35e3a6a..f28dbc2 100644 --- a/lib/sum_data_collector.py +++ b/lib/sum_data_collector.py @@ -133,17 +133,13 @@ def collect_sum_data_from_history(history_id=None, start_date=None, end_date=Non filtered_fields.add('charge_degree') # 充电电量始终采集 sum_results = {k: v for k, v in sum_results.items() if k in filtered_fields} - # 逐条插入求和字段数据 + # 收集所有待插入的数据 + batch_params = [] + + # 收集求和字段数据 for field_key, field_value in sum_results.items(): - sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count - sql = ''' - INSERT INTO t_daily_report_sum_data - (id, report_date, config_id, config_name, split_type, split_value, split_name, - sum_field_key, sum_field_name, sum_value, total_orders, sort_order, data_source, remark, - create_time, update_time) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) - ''' - params = ( + sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count + len(batch_params) + batch_params.append(( sum_data_id, history['report_date'], history['config_id'], @@ -158,11 +154,10 @@ def collect_sum_data_from_history(history_id=None, start_date=None, end_date=Non sort_order, 'auto', f'从历史记录自动采集，历史ID: {history["id"]}' - ) - execute_insert(sql, params) - collected_count += 1 + )) # 如果配置了自定义服务费单价，自动计算并采集自定义服务费 + custom_service_fee = None if custom_service_fee_price is not None and custom_service_fee_price != '' and sum_results.get('charge_degree') is not None: try: charge_degree = float(sum_results['charge_degree']) @@ -171,15 +166,8 @@ def collect_sum_data_from_history(history_id=None, start_date=None, end_date=Non fee_display_name = str(custom_service_fee_name).strip() else: fee_display_name = f'自定义服务费({custom_service_fee_price}元/kWh)' - sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count - sql = ''' - INSERT INTO t_daily_report_sum_data - (id, report_date, config_id, config_name, split_type, split_value, split_name, - sum_field_key, sum_field_name, sum_value, total_orders, sort_order, data_source, remark, - create_time, update_time) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) - ''' - params = ( + sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count + len(batch_params) + batch_params.append(( sum_data_id, history['report_date'], history['config_id'], @@ -194,16 +182,17 @@ def collect_sum_data_from_history(history_id=None, start_date=None, end_date=Non sort_order, 'auto', f'自动计算：充电电量{charge_degree}kWh × 单价{custom_service_fee_price}元/kWh' - ) - execute_insert(sql, params) - collected_count += 1 - - # 计算自定义实收总金额（实收电费 + 自定义服务费） - # 实收电费从 sum_results 中获取（charge_elecfee_amount） + )) + except Exception as fee_error: + log_warning(f"[采集] 历史记录 {history['id']} 自定义服务费计算失败: {fee_error}", 'sum_data') + + # 计算自定义实收总金额（实收电费 + 自定义服务费） + if custom_service_fee is not None: + try: charge_elecfee_amount = float(sum_results.get('charge_elecfee_amount', 0) or 0) custom_total_amount = round(charge_elecfee_amount + custom_service_fee, 2) - sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count - params = ( + sum_data_id = int(datetime.now().timestamp() * 1000000) + collected_count + len(batch_params) + batch_params.append(( sum_data_id, history['report_date'], history['config_id'], @@ -218,11 +207,22 @@ def collect_sum_data_from_history(history_id=None, start_date=None, end_date=Non sort_order, 'auto', f'自动计算：实收电费{charge_elecfee_amount}元 + 自定义服务费{custom_service_fee}元' - ) - execute_insert(sql, params) - collected_count += 1 - except Exception as fee_error: - log_error(f"[采集] 计算自定义服务费失败: {fee_error}", 'sum_data') + )) + except Exception as total_error: + log_warning(f"[采集] 历史记录 {history['id']} 自定义实收总金额计算失败: {total_error}", 'sum_data') + + # 批量插入所有数据（Doris 优化） + if batch_params: + sql = ''' + INSERT INTO t_daily_report_sum_data + (id, report_date, config_id, config_name, split_type, split_value, split_name, + sum_field_key, sum_field_name, sum_value, total_orders, sort_order, data_source, remark, + create_time, update_time) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) + ''' + from lib.db import execute_batch_insert + inserted = execute_batch_insert(sql, batch_params, batch_size=50) + collected_count += inserted log_info(f'[采集] 采集完成，共采集 {collected_count} 条求和数据，跳过 {skipped_count} 条无求和数据的记录', 'sum_data') return { @@ -289,7 +289,7 @@ def _calculate_charge_degree(history): def get_sum_data_list(page=None, page_size=20, start_date=None, end_date=None, config_id=None, - config_name=None, sum_field_key=None, data_source=None): + config_name=None, sum_field_key=None, data_source=None, skip_count=False): """ 查询求和数据列表 @@ -302,6 +302,7 @@ def get_sum_data_list(page=None, page_size=20, start_date=None, end_date=None, c config_name: 配置名称（可选，模糊搜索） sum_field_key: 求和字段键名（可选） data_source: 数据来源（可选） + skip_count: 是否跳过 COUNT 查询（优化性能） Returns: dict: 查询结果 @@ -336,12 +337,14 @@ def get_sum_data_list(page=None, page_size=20, start_date=None, end_date=None, c where_sql = ' AND '.join(where_clauses) - # 查询总数 - total_result = execute_query( - f'SELECT COUNT(*) as total FROM t_daily_report_sum_data WHERE {where_sql}', - tuple(params) - ) - total = total_result[0]['total'] + # 查询总数（如果不跳过） + total = 0 + if not skip_count: + total_result = execute_query( + f'SELECT COUNT(*) as total FROM t_daily_report_sum_data WHERE {where_sql}', + tuple(params) + ) + total = total_result[0]['total'] # 查询数据 if page is not None: