62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
#!/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())
|