diff --git a/models/dm_model_perf_daily.json b/models/dm_model_perf_daily.json index 95dd9ea..55612fe 100644 --- a/models/dm_model_perf_daily.json +++ b/models/dm_model_perf_daily.json @@ -38,7 +38,7 @@ "name": "catelogid", "title": "目录ID", "type": "str", - "length": 10 + "length": 32 }, { "name": "providerid", diff --git a/sage_datamart/etl.py b/sage_datamart/etl.py index 2f9698a..31c1adf 100644 --- a/sage_datamart/etl.py +++ b/sage_datamart/etl.py @@ -178,7 +178,7 @@ async def aggregate_daily_perf(sor, stat_date=None): SUM(amount) as total_amount FROM dm_model_call_fact WHERE call_date = ${stat_date}$ - GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, userorgid, dm_model_call_fact.currency + GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, dm_model_call_fact.userorgid, dm_model_call_fact.currency """ rows = await sor.sqlExe(sql, {'stat_date': stat_date}) info('sage_datamart AGG: aggregating ' + str(len(rows)) + ' groups for ' + stat_date) diff --git a/scripts/aggregate_all.py b/scripts/aggregate_all.py new file mode 100644 index 0000000..92febbd --- /dev/null +++ b/scripts/aggregate_all.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +""" +数据集市聚合脚本 — 从 dm_model_call_fact 重算所有历史日期的聚合表 +独立于 init_data.py,可单独执行。 + +用法: cd /path/to/sage && python /path/to/sage_datamart/scripts/aggregate_all.py +""" +import asyncio +from datetime import datetime +from appPublic.jsonConfig import getConfig +from appPublic.uniqueID import getID +from appPublic.dictObject import DictObject +from sqlor.dbpools import DBPools + +DB_NAME = 'sage' + + +async def batch_aggregate_history(sor): + """按 dm_model_call_fact 中所有历史日期重算聚合表""" + from sage_datamart.etl import aggregate_daily_perf, aggregate_provider_cost + + rows = await sor.sqlExe( + 'SELECT DISTINCT call_date FROM dm_model_call_fact ORDER BY call_date', {} + ) + dates = [] + for r in rows: + if hasattr(r, 'call_date'): + d = r.call_date + else: + d = r.get('call_date', r[0]) + dates.append(d.strftime('%Y-%m-%d') if hasattr(d, 'strftime') else str(d)) + + print('Aggregating ' + str(len(dates)) + ' distinct dates...') + ok = fail = 0 + for i, d in enumerate(dates): + try: + await aggregate_daily_perf(sor, stat_date=d) + await aggregate_provider_cost(sor, stat_date=d) + ok += 1 + except Exception as e: + fail += 1 + print(' [' + d + '] FAILED: ' + str(e)[:120]) + if (i + 1) % 10 == 0: + print(' progress: ' + str(i + 1) + '/' + str(len(dates))) + + print('Done: ' + str(ok) + ' ok, ' + str(fail) + ' failed') + return ok, fail + + +async def main(): + config = getConfig('.') + db = DBPools() + db.databases = config.databases + async with db.sqlorContext(DB_NAME) as sor: + ok, fail = await batch_aggregate_history(sor) + print('\n=== Aggregation Complete ===') + print('ok=' + str(ok) + ' fail=' + str(fail)) + + +if __name__ == '__main__': + asyncio.run(main())