diff --git a/sage_datamart/etl.py b/sage_datamart/etl.py index e2453d8..9174779 100644 --- a/sage_datamart/etl.py +++ b/sage_datamart/etl.py @@ -1,106 +1,16 @@ """ 数据集市 ETL — 增量同步 llmusage → 集市表 +建表由 models/*.json + json2ddl + build.sh 部署时完成，ETL 只负责数据读写。 """ import json -import asyncio from datetime import datetime, timedelta from appPublic.uniqueID import getID from appPublic.dictObject import DictObject -from appPublic.jsonConfig import getConfig -from appPublic.log import debug, info, exception -from sqlor.dbpools import DBPools -from ahserver.serverenv import ServerEnv +from appPublic.log import info MODULE_NAME = "sage_datamart" -async def ensure_tables(sor): - """创建集市表（如不存在）""" - tables = [ - """CREATE TABLE IF NOT EXISTS dm_model_call_fact ( - id VARCHAR(32) PRIMARY KEY, - luid VARCHAR(32) NOT NULL, - llmid VARCHAR(32) NOT NULL, - model VARCHAR(100), - catelogid VARCHAR(10), - userid VARCHAR(32) NOT NULL, - userorgid VARCHAR(32) NOT NULL, - ownerid VARCHAR(32), - providerid VARCHAR(32), - call_date DATE NOT NULL, - call_hour TINYINT NOT NULL, - call_time DATETIME NOT NULL, - ttft_ms INT, - ttot_ms INT, - prompt_tokens INT DEFAULT 0, - completion_tokens INT DEFAULT 0, - status VARCHAR(20), - fail_reason VARCHAR(200), - amount DECIMAL(10,4) DEFAULT 0, - currency VARCHAR(5) DEFAULT 'CNY', - distributor_orgid VARCHAR(32), - sale_userid VARCHAR(32), - created_at DATETIME DEFAULT NOW(), - INDEX idx_call_date (call_date), - INDEX idx_llmid (llmid), - INDEX idx_userorgid (userorgid), - INDEX idx_ownerid (ownerid), - INDEX idx_status (status), - INDEX idx_model (model) - )""", - - """CREATE TABLE IF NOT EXISTS dm_model_perf_daily ( - id VARCHAR(32) PRIMARY KEY, - stat_date DATE NOT NULL, - llmid VARCHAR(32) NOT NULL, - model VARCHAR(100), - catelogid VARCHAR(10), - providerid VARCHAR(32), - ownerid VARCHAR(32), - total_calls INT DEFAULT 0, - success_calls INT DEFAULT 0, - fail_calls INT DEFAULT 0, - success_rate DECIMAL(5,2), - avg_ttft_ms DECIMAL(10,1), - p50_ttft_ms DECIMAL(10,1), - p95_ttft_ms DECIMAL(10,1), - p99_ttft_ms DECIMAL(10,1), - avg_ttot_ms DECIMAL(10,1), - p50_ttot_ms DECIMAL(10,1), - p95_ttot_ms DECIMAL(10,1), - max_concurrent INT, - avg_concurrent DECIMAL(5,1), - total_prompt_tokens BIGINT DEFAULT 0, - total_completion_tokens BIGINT DEFAULT 0, - total_amount DECIMAL(12,4) DEFAULT 0, - userorgid VARCHAR(32), - distributor_orgid VARCHAR(32), - UNIQUE KEY uk_date_model_org (stat_date, llmid, userorgid), - INDEX idx_date (stat_date), - INDEX idx_model (model) - )""", - - """CREATE TABLE IF NOT EXISTS dm_provider_cost_daily ( - id VARCHAR(32) PRIMARY KEY, - stat_date DATE NOT NULL, - model VARCHAR(100) NOT NULL, - catelogid VARCHAR(10), - providerid VARCHAR(32) NOT NULL, - provider_name VARCHAR(100), - total_calls INT DEFAULT 0, - total_tokens BIGINT DEFAULT 0, - total_amount DECIMAL(12,4) DEFAULT 0, - unit_price DECIMAL(10,6), - avg_ttft_ms DECIMAL(10,1), - UNIQUE KEY uk_date_model_provider (stat_date, model, providerid), - INDEX idx_model (model), - INDEX idx_date (stat_date) - )""", - ] - for sql in tables: - await sor.sqlExe(sql, {}) - - def parse_usages(usages_str): """从 llmusage.usages JSON 中提取 token 数""" try: @@ -122,7 +32,9 @@ async def sync_call_fact(sor, last_sync=None): last_sync = (datetime.now() - timedelta(minutes=10)).strftime('%Y-%m-%d %H:%M:%S') sql = """ - SELECT lu.*, l.model as llm_model, l.catelogid, l.providerid, l.ownerid + SELECT lu.*, l.model as llm_model, l.providerid, + (SELECT llmcatelogid FROM llm_api_map + WHERE llmid = lu.llmid AND isdefaultcatelog = '1' LIMIT 1) as catelogid FROM llmusage lu JOIN llm l ON l.id = lu.llmid WHERE lu.use_time >= ${last_sync}$ @@ -163,7 +75,7 @@ async def sync_call_fact(sor, last_sync=None): 'completion_tokens': completion_tokens, 'status': r.status or 'SUCCEEDED', 'amount': r.amount or 0, - 'currency': getattr(r, 'currency', 'CNY') or 'CNY', + 'currency': getattr(r, 'amount_currency', None) or 'CNY', } await sor.C('dm_model_call_fact', ns) count += 1 @@ -269,18 +181,15 @@ async def aggregate_provider_cost(sor, stat_date=None): async def run_etl_sync(sor): - """每5分钟运行: 确保表存在 + 增量同步""" - await ensure_tables(sor) + """每5分钟运行: 增量同步""" await sync_call_fact(sor) async def run_etl_aggregate(sor): """每小时运行: 聚合天级性能 + 并发统计""" - await ensure_tables(sor) await aggregate_daily_perf(sor) async def run_etl_provider_cost(sor): """每天运行: 供应商性价比""" - await ensure_tables(sor) await aggregate_provider_cost(sor) diff --git a/scripts/init_data.py b/scripts/init_data.py index ee08f87..f9065d4 100644 --- a/scripts/init_data.py +++ b/scripts/init_data.py @@ -1,12 +1,12 @@ #!/usr/bin/env python3 """ 数据集市初始化脚本 — 从 llmusage + llmusage_history 导入全量历史数据到 dm_* +建表由 models/*.json + json2ddl + build.sh 部署时完成，本脚本只负责数据读写。 用法: cd /path/to/sage && python /path/to/sage_datamart/scripts/init_data.py """ import asyncio import json -import sys from datetime import datetime from sqlor.dbpools import DBPools from appPublic.jsonConfig import getConfig @@ -26,7 +26,7 @@ def parse_usages(usages_str): u.get('completion_tokens', 0) or 0, u.get('total_tokens', 0) or 0 ) - except: + except Exception: pass return 0, 0, 0 @@ -42,7 +42,9 @@ async def batch_sync_call_fact(sor, source_table, batch_size=1000): while True: sql = ( - 'SELECT lu.*, l.model as llm_model, l.catelogid, l.providerid, l.ownerid ' + 'SELECT lu.*, l.model as llm_model, l.providerid, ' + '(SELECT llmcatelogid FROM llm_api_map ' + 'WHERE llmid = lu.llmid AND isdefaultcatelog = \'1\' LIMIT 1) as catelogid ' 'FROM ' + source_table + ' lu ' 'JOIN llm l ON l.id = lu.llmid ' 'ORDER BY lu.use_time ' @@ -75,8 +77,8 @@ async def batch_sync_call_fact(sor, source_table, batch_size=1000): 'id': getID(), 'luid': r.id, 'llmid': r.llmid, - 'model': r.llm_model or r.model, - 'catelogid': r.catelogid, + 'model': r.llm_model, + 'catelogid': r.catelogid or '', 'userid': r.userid, 'userorgid': r.userorgid, 'ownerid': r.ownerid, @@ -90,7 +92,7 @@ async def batch_sync_call_fact(sor, source_table, batch_size=1000): 'completion_tokens': completion_tokens, 'status': r.status or 'SUCCEEDED', 'amount': r.amount or 0, - 'currency': getattr(r, 'currency', 'CNY') or 'CNY', + 'currency': getattr(r, 'amount_currency', None) or 'CNY', } await sor.C('dm_model_call_fact', ns) total_inserted += 1 @@ -122,23 +124,17 @@ async def batch_aggregate_history(sor): async def run_init(sor): """ 初始化流程: - 1. 建表（如不存在） - 2. 从 llmusage_history 批量导入历史 - 3. 从 llmusage 导入当前数据 - 4. 重算所有日期的聚合表 + 1. 从 llmusage_history 批量导入历史 + 2. 从 llmusage 导入当前数据 + 3. 重算所有日期的聚合表 """ - from sage_datamart.etl import ensure_tables - - print('[1/4] Creating tables...') - await ensure_tables(sor) - - print('[2/4] Importing llmusage_history...') + print('[1/3] Importing llmusage_history...') ins_hist, skip_hist = await batch_sync_call_fact(sor, 'llmusage_history') - print('[3/4] Importing llmusage...') + print('[2/3] Importing llmusage...') ins_curr, skip_curr = await batch_sync_call_fact(sor, 'llmusage') - print('[4/4] Aggregating historical dates...') + print('[3/3] Aggregating historical dates...') dates = await batch_aggregate_history(sor) return {