refactor: remove DDL from ETL/init, use llm_api_map for catelogid
- Remove ensure_tables() from etl.py (DDL belongs in models/*.json + build.sh) - Use llm_api_map.isdefaultcatelog='1' to get catelogid (llm table no longer has catelogid) - Fix currency field: use amount_currency from llmusage_history - init_data.py: remove ensure_tables import and call, reduce to 3 steps
This commit is contained in:
parent
c78f3be577
commit
e84fc8882b
@ -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)
|
||||
|
||||
@ -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 {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user