#!/usr/bin/env python # -*- coding: utf-8 -*- """ ODS 层原始表同步任务 说明：轻量级增量同步，只同步昨天的数据，对 MySQL 友好 """ import datetime from db.pool import get_mysql_conn, get_doris_conn from logger.handler import get_logger logger = get_logger("job.ods_patient_exam_sync") def run(start_date=None, end_date=None): """执行 ODS 层原始表同步""" if start_date is None or end_date is None: yesterday = datetime.date.today() - datetime.timedelta(days=1) start_date = end_date = yesterday.strftime("%Y-%m-%d") logger.info("=" * 60) logger.info(f"[ODS同步] 开始同步: {start_date}") logger.info("=" * 60) # 定义需要同步的表 tables = [ { "name": "ods_patient_custome", "source_table": "patient_custome", "time_field": "update_time", "sync_mode": "increment", # 增量 "columns": """ id, patientId, name, sex, birthday, identity, mobile, source, inviter, inviter_phone, customer_optometrist_staff_id, create_time, update_time """ }, { "name": "ods_clinic_clinic", "source_table": "clinic_clinic", "time_field": "create_time", "sync_mode": "increment", "columns": """ id, patient_id, org_id, type, channel, add_user, appoint_date, AppointmentTime_start, matter, accept_doctor_id, accept_optometrist_id, accept_assistant_id, come_time, come_date, reg_status, pno, create_time, update_time """ }, { "name": "ods_exam_examresult", "source_table": "exam_examresult", "time_field": "created", "sync_mode": "increment", "columns": """ id, clinic_id, org_id, result, tab, visual, exam_id, created, create_time, update_time """ }, { "name": "ods_treat_sellgoods", "source_table": "treat_sellgoods", "time_field": "create_time", "sync_mode": "increment", "columns": """ id, sell_goods_list_id, goods_id, out_time, out_date, remark, create_time, update_time """ }, { "name": "ods_material_goods", "source_table": "material_goods", "time_field": None, "sync_mode": "full", # 全量（数据量小） "columns": """ id, name, uuid_no, fee_class, group_id, create_time, update_time """ }, { "name": "ods_fee_order", "source_table": "fee_order", "time_field": "create_time", "sync_mode": "increment", "columns": """ id, patientId, org_id, pay_status, pay_time, pay_date, total_amt, create_time, update_time """ }, { "name": "ods_fee_orderdetail", "source_table": "fee_orderdetail", "time_field": "create_time", "sync_mode": "increment", "columns": """ id, order_id, goods_id, patient_id, name, pay_time, pay_date, price, quantity, total_amt, create_time, update_time """ }, { "name": "ods_staff_staff", "source_table": "staff_staff", "time_field": None, "sync_mode": "full", # 全量（数据量小） "columns": """ id, user_id, main_org_id, name, mobile, qywx_userid, status, create_time, update_time """ }, { "name": "ods_org_org", "source_table": "org_org", "time_field": None, "sync_mode": "full", # 全量（数据量小） "columns": """ id, name, group_id, is_effective, create_time, update_time """ }, ] mysql_conn = get_mysql_conn() doris_conn = get_doris_conn() try: success_count = 0 for table_config in tables: try: count = sync_table( mysql_conn, doris_conn, table_config, start_date, end_date ) if count > 0: success_count += 1 except Exception as e: logger.error(f"同步 {table_config['name']} 失败: {e}") import traceback traceback.print_exc() logger.info("=" * 60) logger.info(f"[ODS同步] 同步完成，成功 {success_count}/{len(tables)} 个表") logger.info("=" * 60) return success_count finally: mysql_conn.close() doris_conn.close() def sync_table(mysql_conn, doris_conn, table_config, start_date, end_date): """同步单个表""" table_name = table_config["name"] source_table = table_config["source_table"] time_field = table_config.get("time_field") sync_mode = table_config["sync_mode"] columns = table_config["columns"] logger.info(f"\n开始同步: {table_name} ({sync_mode})") mysql_cursor = mysql_conn.cursor() doris_cursor = doris_conn.cursor() try: # 1. 从 MySQL 查询数据 if sync_mode == "increment" and time_field: # 增量同步：只查询昨天的数据 sql = f""" SELECT {columns} FROM {source_table} WHERE {time_field} >= %s AND {time_field} < %s """ params = [f"{start_date} 00:00:00", f"{end_date} 23:59:59"] else: # 全量同步 sql = f"SELECT {columns} FROM {source_table}" params = [] logger.info(f"执行SQL: {sql}") logger.info(f"参数: {params}") mysql_cursor.execute(sql, params) data_list = mysql_cursor.fetchall() row_count = len(data_list) logger.info(f"查询到 {row_count} 条记录") if row_count == 0: logger.info(f"无新数据，跳过") return 0 # 2. 构建插入 SQL # 获取列名 columns_list = [col.strip() for col in columns.split(',')] # 添加源表时间字段 if sync_mode == "increment" and time_field: columns_list.extend(["source_create_time", "source_update_time"]) placeholders = ', '.join(['%s'] * len(columns_list)) insert_sql = f"INSERT INTO {table_name} ({', '.join(columns_list)}) VALUES ({placeholders})" # 3. 准备数据 values_list = [] for row in data_list: row_dict = dict(row) # 转换数据 values = [] for col in columns_list: if col in ["source_create_time", "source_update_time"]: # 源表时间字段 val = row_dict.get(time_field) else: # 普通字段 val = row_dict.get(col) # 处理日期时间 if val and col.endswith("_time"): if isinstance(val, str): pass # 已经是字符串 elif hasattr(val, 'strftime'): val = val.strftime('%Y-%m-%d %H:%M:%S') else: val = None elif val and col.endswith("_date"): if isinstance(val, str): pass # 已经是字符串 elif hasattr(val, 'strftime'): val = val.strftime('%Y-%m-%d') else: val = None values.append(val) values_list.append(tuple(values)) # 4. 批量插入 Doris batch_size = 100 total_inserted = 0 for i in range(0, len(values_list), batch_size): batch = values_list[i:i+batch_size] doris_cursor.executemany(insert_sql, batch) doris_conn.commit() total_inserted += len(batch) logger.info(f"已写入 {total_inserted}/{row_count}") logger.info(f"✓ {table_name} 同步完成: {total_inserted} 条") return total_inserted except Exception as e: logger.error(f"同步 {table_name} 失败: {e}") raise finally: mysql_cursor.close() doris_cursor.close()