5.2 创建影子DB

即在备份DB创建多个库的影子库（精确到月），以及里面的表（精确到天）。

5.3 dump同步数据

dump生产数据，使用create_time对数据进行筛选。然后再备份库里面重放。

5.4 diff 核对数据

核对源库和1.2 脚本dump的备份库的数据。

• 源库和备份库都执行dump(相同的时间区间)。
• 使用diff校验2个sql文件是否相同。

8.2 开发调试参考步骤

• 在个人分支开发后，push
• 确定部署环境（寻找目前无人使用的机器）
• 确定数据来源（具体db）
• 检查数据正确性
  • 确认数据是否完整，主要体现在⼀个订单的整个⽣命周期是否正常。
  • 确认数据有自己修改的特性
• 提单部署
• 运行
• 查看日志和生成的报表是否正常

8.3 发布参考步骤

• 在个人分支开发后，先在UAT环境测试新增特性
• 提MR将个人分支合入test分支
• 提单将test分支代码合入master，填写《checklist》、《发布审核》
• 将master版本，部署到gamma环境。
  • 对比prod和gamma环境flow_new表的数据总数。
  • 检查两环境各个报表的md5是否相同
• 按照验证计划开始验证prodtest
  • 检查发布特性
  • 对比生产环境、重放环境生成的文件
  • 对比生成耗时，prod、gamma对比
  • 同步填写《交易报表prod和gamma数据准确性报告》
• 将master版本，部署到prod环境。
• 按照验证计划开始验证prod​​​​​​​

十、性能

随着业务规模的不断增长，数据库实例和表的数量也逐渐增加，串行 dump 方法因逐条顺序执行，效率低下，无法满足业务对数据备份和导出的时效性要求。

因此，我们必须开发一个支持并发控制的脚本，通过限制最大并发任务数，在提升执行效率的同时，确保资源的合理利用和系统的稳定。

并发控制

1. 存储计数器：在 Redis 或 MySQL 中维护一个计数器，用来记录当前正在运行的 dump 脚本数量。
2. 并发控制逻辑：
   • 每个脚本启动时，检查计数器是否小于允许的最大并发数。
   • 如果小于，增加计数并继续执行脚本。
   • 如果达到上限，脚本等待或直接退出。
3. 计数清理：在脚本完成后，减少计数器，释放占用的并发槽位。

需要注意的点：

关键点说明

1. Redis 计数器

   • 使用 redis.incr 和 redis.decr 操作维护当前正在执行任务的计数。
   • 可以通过 redis.get(redis_key) 随时查看当前计数。

2. 并发限制

   • 在 acquire_lock 中检查计数是否小于 MAX_CONCURRENT_TASKS，满足条件时增加计数。
   • 如果达到限制，可以选择等待或直接跳过（可扩展为重试机制）。

3. 脚本结束

   • 无论执行成功还是失败，都在 finally 块中调用 release_lock，确保计数器准确。

4. 并发安全性

   • Redis 操作是原子性的，确保多个脚本同时运行时，计数器不会出现竞争问题。

我提供一个实例代码，当然，你需要了解原理后自己写一个定制化功能

import redis
import time
import subprocess

# 配置 Redis
redis_key = "dump_task_count"
MAX_CONCURRENT_TASKS = 5
TIMEOUT_THRESHOLD = 300  # 超时时间（秒）

# 初始化 Redis 客户端
redis_client = redis.StrictRedis(decode_responses=True)

def acquire_lock():
    """尝试获取执行权限并记录开始时间"""
    task_data = redis_client.get(redis_key)
    if task_data:
        current_count, start_time = map(int, task_data.split(":"))
    else:
        current_count, start_time = 0, 0

    if current_count < MAX_CONCURRENT_TASKS:
        # 更新计数器和开始时间
        redis_client.set(redis_key, f"{current_count + 1}:{int(time.time())}")
        return True
    return False

def release_lock():
    """释放执行权限并检查是否超时"""
    task_data = redis_client.get(redis_key)
    if not task_data:
        return  # 没有记录，无需释放

    current_count, start_time = map(int, task_data.split(":"))
    current_time = int(time.time())
    if current_count > 0:
        redis_client.set(redis_key, f"{current_count - 1}:{start_time}")

    # 检查是否超时
    if current_time - start_time > TIMEOUT_THRESHOLD:
        print(f"ALERT: Task exceeded timeout threshold! Execution time: {current_time - start_time} seconds.")

def run_dump_command(command):
    """执行 dump 脚本并控制并发"""
    if not acquire_lock():
        print(f"Max concurrent tasks reached. Skipping: {command}")
        return False

    try:
        print(f"Executing: {command}")
        subprocess.run(command, shell=True, check=True)
        print(f"Completed: {command}")
    except subprocess.CalledProcessError as e:
        print(f"Error occurred: {e}")
    finally:
        release_lock()  # 无论成功或失败都释放锁
    return True

# 示例 dump 命令
dump_commands = [
    "mysqldump -h host1 -u user -p'password' db1 > dump1.sql",
    "mysqldump -h host2 -u user -p'password' db2 > dump2.sql",
    "mysqldump -h host3 -u user -p'password' db3 > dump3.sql",
    "mysqldump -h host4 -u user -p'password' db4 > dump4.sql",
    "mysqldump -h host5 -u user -p'password' db5 > dump5.sql",
    "mysqldump -h host6 -u user -p'password' db6 > dump6.sql"
]

# 执行 dump 脚本
for command in dump_commands:
    run_dump_command(command)

在第一二阶段时，并发控制可以显著提高效率，但是请你注意DB配置，如果并发太高，DB很可能扛不住挂掉，上线前请做压测。

但是在三阶段方案，只有控制并发就不够了，因为这个阶段需要处理所有订单，每个服务实例必须知道自己需要处理哪部分订单

订单分配

一致性哈希不想过多介绍，可以处理这种场景，还可以载流量徒增的时候动态调整。

在本问题中：

• 数据分布：通过一致性哈希，将订单按照 order_id 分配到不同的处理节点（机器）。
• 并发处理：每个节点负责处理自己分配的订单，独立完成补填并写入新表，减少单点压力。

示例代码：

import hashlib
from collections import defaultdict

# 模拟节点列表
nodes = ["Machine-1", "Machine-2", "Machine-3", "Machine-4"]

# 模拟数据表中订单ID
orders = [1001, 1002, 1003, 1004, 2001, 2002, 2003, 3001, 4001, 5001]

def consistent_hash(key, node_list):
    """基于一致性哈希将key分配到某个节点"""
    hash_value = int(hashlib.md5(str(key).encode()).hexdigest(), 16)
    index = hash_value % len(node_list)
    return node_list[index]

# 将订单分配到节点
node_to_orders = defaultdict(list)
for order_id in orders:
    assigned_node = consistent_hash(order_id, nodes)
    node_to_orders[assigned_node].append(order_id)

# 输出分配结果
for node, assigned_orders in node_to_orders.items():
    print(f"{node} handles orders: {assigned_orders}")

懒人版本

当然，这种方法稍微有些复杂，有些同学比较笨或懒，也不要求动态配置，那再写个简单版本，读配置获知自己的机器需要处理0-9结尾的哪些订单，然后处理

# 配置：当前机器负责的订单尾号
MY_TAILS = {0, 1, 2}  # 假设当前机器负责尾号为0, 1, 2的订单

# 模拟数据表中的订单ID
orders = [1001, 1002, 1003, 1004, 2001, 2002, 2003, 3001, 4001, 5001]

def filter_orders_by_tail(orders, tails):
    """根据尾号筛选属于当前机器的订单"""
    return [order for order in orders if order % 10 in tails]

def process_order(order_id):
    """处理订单逻辑"""
    print(f"Processing order: {order_id}")

# 筛选当前机器需要处理的订单
my_orders = filter_orders_by_tail(orders, MY_TAILS)

# 处理订单
for order in my_orders:
    process_order(order)

十一、正确性的保证

性能问题解决后，上报的数据有没有漏报错报也非常重要，关系着交易链路权限是否会被收回。

1. 字段完整性与格式校验

• 字段完整性：
  • 检查所有必填字段是否有值，例如 ORDER_ID、SOURCE_ORDER_ID、TRADE_ID 等关键字段。
• 字段格式：
  • 校验字段格式是否正确，例如：
    • ORDER_ID：符合指定长度与字符规则。
    • 日期字段：格式应为 YYYY-MM-DD HH:MM:SS。
    • 数值字段：是否为正数，且精度符合预期（如价格精确到小数点后两位）。

---

2. 同一订单的多条记录先后关系校验

多条记录的时间先后关系是必检内容，需确保满足以下逻辑：

• O.RO 依赖 O.NO：
  • 报单响应（O.RO）必须在报单创建（O.NO）之后。
• T 依赖 O.RO：
  • 成交（T）必须在报单响应（O.RO）之后。
• A.RJ 依赖 O.NO：
  • 拒绝操作（A.RJ）必须在报单创建（O.NO）之后。
• A.C 依赖 O.NO：
  • 取消操作（A.C）必须在报单创建（O.NO）之后。
• M 依赖 T：
  • 成交修改（M）必须发生在成交（T）之后。

实现方式：

• 按 ORDER_ID 对记录排序，并逐条检查时间戳是否符合上述规则。
• 标记不满足时间顺序的记录，供进一步分析。

---

3. SOURCE_ORDER_ID 校验

• 有效性校验：
  • SOURCE_ORDER_ID 必须存在并有效，关联的上游订单必须能够在系统中找到。
• 正确性校验：
  • 确认 SOURCE_ORDER_ID 对应的订单状态是否合理，如订单被取消或已完成的情况下不应再出现新的成交记录。

---

4. Overfill（超量成交）校验

• 定义：
  • 单个订单的成交量总和不应超过订单的委托量。
• 校验方式：
  • 对每个 ORDER_ID 汇总成交量，检查是否超出 ORDER_QTY（委托量）。

SELECT ORDER_ID, SUM(TRADE_QTY) AS TOTAL_QTY, ORDER_QTY 
FROM trade_table 
GROUP BY ORDER_ID HAVING TOTAL_QTY > ORDER_QTY;

---

5. 数据统计与汇总校验

对报表的核心统计数据进行汇总校验，确保整体一致性：

• 总成交额：
  • 计算所有成交记录的总成交金额（SUM(TRADE_QTY * TRADE_PRICE)），与外部对账系统的数据对比。
• 总成交股数：
  • 汇总所有成交记录的股数（SUM(TRADE_QTY)）。
• 总成交笔数：
  • 统计成交记录的总条数。
• 总订单数量：
  • 统计订单记录的总条数，与委托数量对比，确保订单未丢失。

---

6. 数据关联性校验

• 跨表校验：
  • 确保订单、成交、修改记录间的关联关系正确。例如：
    • TRADE_ID 必须关联到有效的 ORDER_ID。
    • 修改记录的 PARENT_ID 必须能匹配到上游订单或成交记录。
• 引用关系检查：
  • 确认所有引用的 ID（如 PARENT_ORDER_ID）是否存在并有效。

---

7. 状态流转校验

• 状态变化合理性：
  • 确保订单状态的流转符合业务逻辑。例如：
    • NEW -> PARTIAL_FILL -> FILLED -> COMPLETED
    • 状态不能逆流（如 FILLED 后又变为 PARTIAL_FILL）。
• 最终状态检查：
  • 确认所有订单的状态都能流转到合理的终态（如 COMPLETED 或 CANCELED）。

---

8. 重复数据校验

• 订单重复：
  • 检查 ORDER_ID 是否重复，确保同一订单号不出现多条重复的报单记录。
• 成交重复：
  • 检查 TRADE_ID 是否重复，避免同一成交记录被多次写入。

---

9. 数据异常校验

• 空值校验：
  • 确保关键字段无空值，例如 ORDER_ID、TRADE_ID、TRADE_QTY 等。
• 异常值校验：
  • 确保成交量、成交金额为正值，不存在负数或异常数据。
• 逻辑校验：
  • 确保成交量与价格的乘积等于成交金额（精度误差除外）。

---

10. 对账结果记录

• 生成对账报告：
  • 将所有校验结果记录在对账报告中，包括：
    • 每项校验的通过与失败情况。
    • 错误记录的详细信息（如订单号、错误类型、字段值）。
• 自动告警：
  • 对严重错误（如漏报、错报）触发告警，通知相关人员及时处理。

---

通过以上对账逻辑，可以有效发现报表生成中的数据异常和业务逻辑问题，确保数据的完整性和准确性。当然，这些也就是我的一家之言，还有很多不完善的地方，请你根据自己的业务来增删改查规则。

十二、Q&A

我还是不知道分段是怎么分的，文章在这一块提的不多。

# 调用主脚本，每30分钟运行一次，覆盖盘前、盘中、盘后
0 9-23 * * * python3 report_pipeline.py --start_time "$(date -d '30 minutes ago' '+%Y-%m-%d %H:%M:00')" --end_time "$(date '+%Y-%m-%d %H:%M:00')"
30 9-22 * * * python3 report_pipeline.py --start_time "$(date -d '30 minutes ago' '+%Y-%m-%d %H:%M:00')" --end_time "$(date '+%Y-%m-%d %H:%M:00')"

假设 report_pipeline.py 包含以下功能：

1. 数据 dump：从数据库中按时间段导出原始数据。
2. 数据汇聚：处理数据，按订单维度补充信息。
3. 报表生成：从汇聚表中读取数据并生成报表文件。

每次调用脚本时，将基于 --start_time 和 --end_time 参数处理对应时间段的数据。

脚本运行时间表（UTC）

时间点	说明
09:00	处理盘前的第一段数据
09:30	处理盘前的下一段数据
...
14:30	开始处理盘中数据
21:00	开始处理盘后数据
23:30	处理最后一段盘后数据