# -*- coding: utf-8 -*- import os import sys import traceback root_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__))) if root_dir not in sys.path: sys.path.append(root_dir) print("******** sys.path ********") print(sys.path) from multiprocessing import Process from odps import ODPS from threading import Timer import threading from my_utils import RedisHelper, execute_sql_from_odps from my_config import set_config from log import Log import json from datetime import datetime from queue import Queue from tqdm import tqdm import time config_, _ = set_config() log_ = Log() redis_helper = RedisHelper() REDIS_PREFIX = "redis:vid_vovh24pred_time:" EXPIRE_TIME = 3 * 3600 global PARTITION def worker(queue, executor): while True: row = queue.get() if row is None: # 结束信号 queue.task_done() break executor(row) queue.task_done() def records_process_for_list(records, executor, max_size=50, num_workers=10): # 创建一个线程安全的队列 queue = Queue(maxsize=max_size) # 可以调整 maxsize 以控制内存使用 # 设置线程池大小 num_workers = num_workers # 启动工作线程 threads = [] for _ in range(num_workers): t = threading.Thread(target=worker, args=(queue, executor)) t.start() threads.append(t) # 读取数据并放入队列 for row in tqdm(records): queue.put(row) # 发送结束信号 for _ in range(num_workers): queue.put(None) # 等待所有任务完成 queue.join() # 等待所有工作线程结束 for t in threads: t.join() def process_and_store(row): table_key, json_str = row key = REDIS_PREFIX + ":" + PARTITION + ":" + str(table_key) expire_time = EXPIRE_TIME print(key + "\t" + json_str) redis_helper.set_data_to_redis(key, json_str, expire_time) def check_data(project, table, date, hour) -> int: """检查数据是否准备好,输出数据条数""" odps = ODPS( access_id=config_.ODPS_CONFIG['ACCESSID'], secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'], project=project, endpoint=config_.ODPS_CONFIG['ENDPOINT'], connect_timeout=3000, read_timeout=500000, pool_maxsize=1000, pool_connections=1000 ) try: t = odps.get_table(name=table) log_.info(f"检查分区是否存在-【 dt={date} hh={hour}】") check_res = t.exist_partition(partition_spec=f'dt={date},hh={hour}') if check_res: sql = f'select * from {project}.{table} where dt = {date} and hh = {hour}' log_.info(sql) with odps.execute_sql(sql=sql).open_reader() as reader: data_count = reader.count else: log_.info("表{}分区{}/{}不存在".format(table, date, hour)) data_count = 0 except Exception as e: log_.error("table:{},date:{},hour:{} no data. return data_count=0,报错原因是:{}".format(table, date, hour, e)) data_count = 0 return data_count def get_sql(project, table, date, hour): sql = ''' SELECT vid ,feature FROM {}.{} WHERE dt = '{}' and hh = '{}' '''.format( project, table, date, hour ) print("sql:" + sql) records = execute_sql_from_odps(project=project, sql=sql) video_list = [] with records.open_reader() as reader: for record in reader: key = record['vid'] m = dict() try: json_str = record['feature'] except Exception as e: json_str = json.dumps(m) log_.error(e) log_.error(traceback.format_exc()) video_list.append([key, json_str]) return video_list def main(): try: date = sys.argv[1] hour = sys.argv[2] except Exception as e: date = datetime.now().strftime('%Y%m%d') hour = datetime.now().hour log_.info("没有读取到参数，采用系统时间：{}".format(e)) log_.info("使用时间参数-日期：{}，小时：{}".format(date, str(hour))) if hour in ["00", "01", "02"]: log_.info(f"hour={hour}不执行，直接返回。") return # 1 判断上游数据表是否生产完成 project = "loghubods" table = "alg_vid_vov_hour" run_flag = True begin_ts = int(time.time()) table_data_cnt = 0 while run_flag: if int(time.time()) - begin_ts >= 60 * 40: log_.info("等待上游数据超过40分钟了，认为失败退出:过了{}秒。".format(int(time.time()) - begin_ts)) sys.exit(1) table_data_cnt = check_data(project, table, date, hour) if table_data_cnt == 0: log_.info("上游数据{}未就绪{}/{}，等待...".format(table, date, hour)) log_.info("等待2分钟") time.sleep(60 * 2) else: run_flag = False log_.info("上游数据就绪，count={}，开始读取数据表".format(table_data_cnt)) # 2 读取数据表 处理特征 video_list = get_sql(project, table, date, hour) # 3 写入redis log_.info("video的数据量:{}".format(len(video_list))) partition = str(date) + str(hour) global PARTITION PARTITION = partition records_process_for_list(video_list, process_and_store, max_size=50, num_workers=8) # 4 定义时间的key 独立写入redis redis_helper.set_data_to_redis( key_name=REDIS_PREFIX + "partition", value=partition, expire_time=EXPIRE_TIME ) if __name__ == '__main__': log_.info("开始执行:" + datetime.now().strftime("%Y-%m-%d %H:%M:%S")) process = Process(target=main) process.start() # 等待子进程完成或超时 timeout = 3600 process.join(timeout=timeout) # 设置超时为3600秒（1小时） if process.is_alive(): print("脚本执行时间超过1小时，执行失败，经过了{}秒。".format(timeout)) process.terminate() # 终止子进程 sys.exit(1) # 直接退出主进程并返回状态码999 else: # 检查子进程的返回值 exit_code = process.exitcode if exit_code != 0: print(f"子进程以状态码 {exit_code} 退出，执行失败。") sys.exit(exit_code) # 将子进程的状态码返回到主进程 else: print("子进程正常结束。") # cd /root/zhangbo/rov-offline # python alg_recsys_feature_08_vidh24predv2_redis.py 20241107 14 """ ！！！！！！！！！！！！！！ -----按小时更新----- 更改字段：table 表名 新增了一个partition作为redis的key拼接 REDIS_PREFIX redis的key EXPIRE_TIME redis的过期时间 两段sql 各种字段 注意分区是否有“分钟” record 各种字段 if hour in ["00"]: 哪些小时不执行 process.join(timeout=3600) 任务超时时间3600 int(time.time()) - begin_ts >= 60*50 任务超时时间3000 """