📚 课程目录第3章多源医疗数据采集工程

3. 多源医疗数据采集工程

医疗数据采集不是“把数据拉过来”这么简单。医院内部同时存在 HIS、LIS、PACS、EMR、心电、监护、病理、体检、医保结算等系统，数据形态从结构化表、半结构化报文到 DICOM 影像、PDF 报告、CSV 导出文件不等；对外还要对接区域平台、第三方检测机构、可穿戴设备与科研数据源。采集工程的核心任务，是在时效性、数据量、源系统压力、网络边界、合规约束五个维度之间做出可落地的取舍，并把“取数”变成可调度、可续传、可重试、可观测的稳定链路。

一、五类采集方式的适用边界

批量抽取适合历史数据初始化、T+1 报表、离线科研数据集。典型做法是通过 JDBC 按时间窗或主键区间分片读取，例如按 patient_id % 16 分 16 个分片并行抽取，每片再按 update_time 切分为 10 万行一个批次，避免长事务拖垮源库。批量抽取必须设置只读账号、限流（如每秒不超过 2000 行）、查询超时与连接池上限，并优先走备库或物化视图。

增量捕获解决“只取变化”的问题。常见实现有三类：基于 update_time 或自增 ID 的水位线轮询、基于数据库日志的 CDC（如 MySQL binlog、Oracle LogMiner、SQL Server CDC）、基于触发器或时间戳表的准实时方案。水位线轮询实现简单，但要求源表有可靠且单调的更新时间字段，且需处理“同一时间戳跨批次”的边界；CDC 延迟低、对源库侵入小，但需要日志权限、位点管理和 DDL 变更应对能力。医疗场景中，患者主索引、就诊、医嘱、检验结果等高频变更表更适合 CDC，而字典表、科室表可用低频轮询。

实时流接入面向监护仪、可穿戴设备、急救车、物联网体征网关等持续产生的时序数据。通常由设备或边缘网关通过 MQTT/Kafka 上报，采集端做协议解析、单位归一、时间戳校准后写入消息队列。关键参数包括：采样频率（如心电 250 Hz、血氧 1 Hz）、乱序容忍窗口（如 5 秒）、去重键（设备号+指标+时间戳）、背压策略。实时流不追求“全量落库”，而应先入 Kafka 等缓冲层，再由消费任务按需落库或计算。

文件摆渡用于跨网闸、跨安全域的数据交换，例如检验科 LIS 导出 CSV、影像科导出 DICOM、第三方检测机构回传 PDF+XML。摆渡链路通常为“源端生成文件→摆渡目录→网闸/单向光闸→目标目录→采集程序扫描入库”。工程上必须约定文件命名规范（如 LIS_20240612_001.csv）、校验文件（.md5 或 .ok）、完成标记与失败隔离目录，避免读到半截文件。

API 对接适合区域平台、互联网医院、第三方服务。需要关注鉴权（OAuth2/API Key）、分页（page/size 或 cursor）、限流（429 重试）、幂等（业务主键去重）、字段映射与错误码处理。API 采集应把“拉取”和“解析入库”解耦，先落地原始报文，再异步解析，便于问题回溯。

二、采集任务调度：从“定时跑”到“依赖驱动”

采集任务调度要回答三个问题：什么时候跑、跑什么、跑失败了怎么办。简单场景可用 Cron 表达式，例如 0 30 2 * * ? 表示每天凌晨 2:30 执行 HIS 增量抽取。但医疗数据链路往往存在依赖：患者主索引必须先于就诊、就诊必须先于医嘱、医嘱必须先于计费。此时应使用 DAG 调度（如 Airflow、DolphinScheduler）表达任务依赖，并设置上游成功才触发下游。

调度参数需要显式化：并发度（同一数据源不超过 3 个任务）、超时时间（批量任务 30 分钟、API 任务 5 分钟）、重试次数（默认 3 次）、重试间隔（指数退避 30s/90s/270s）、优先级（实时流 > 增量 > 批量）。对于实时流，调度不是“定时启动”，而是“常驻消费 + 分区分配 + 位点提交”，其可靠性由消费组和 offset 管理保证。

三、断点续传：让失败不从头再来

断点续传的本质是把“进度”持久化。不同采集方式对应不同的进度载体：

批量抽取：按分片记录已完成的主键区间或行号，例如 shard=7, last_id=1048576，重启后从 last_id+1 继续。
增量捕获：CDC 记录 binlog 位点（file+position）或 GTID；水位线轮询记录 last_update_time 与已处理主键集合，防止同一时间戳重复。
实时流：Kafka 消费组按分区提交 offset，重启后从已提交位点继续；若需精确一次，可结合幂等写入或事务表。
文件摆渡：记录已处理文件名与 MD5，处理成功后移动到 done/ 目录，失败移入 error/ 并保留原文件。
API 对接：记录分页游标或最后成功页码，配合业务主键幂等写入，避免重复。

断点续传的检查点必须与业务写入在同一事务或至少保证“先写数据、后更新位点”。若先更新位点再写数据，一旦写数据失败，重启后会跳过该批次，造成数据丢失。

四、失败重试与异常处理

重试不是“无脑重跑”。需要区分可重试错误与不可重试错误：网络超时、连接中断、限流 429、临时锁等待属于可重试；字段类型不匹配、主键冲突、权限不足、文件格式错误属于不可重试，应直接进入死信队列并告警。重试策略建议采用指数退避加抖动，例如第 1 次 30 秒、第 2 次 90 秒、第 3 次 270 秒，并设置最大重试次数，超过后写入 collect_dead_letter 表，记录任务 ID、数据源、错误码、错误信息、原始数据片段和发生时间。

以下是一个采集任务配置示例，用于说明关键参数如何落地：

task:
  name: his_patient_increment
  source: mysql://his-ro:3306/his
  mode: incremental
  watermark_field: update_time
  checkpoint:
    store: redis
    key: collect:his_patient:watermark
  schedule:
    cron: "0 */10 * * * ?"
    timeout_seconds: 300
  retry:
    max_attempts: 3
    backoff: [30, 90, 270]
    retryable_errors: [timeout, connection_reset, http_429]
  sink:
    type: kafka
    topic: ods_his_patient
    idempotent_key: patient_id + update_time

运行过程中要观察四类指标：采集延迟（源端变更到入湖的时间差）、积压量（Kafka lag 或待处理文件数）、成功率（成功批次/总批次）、数据量波动（与历史同期对比，突增突降都需排查）。当延迟持续超过阈值（如增量任务超过 15 分钟）、积压持续增长或成功率低于 99% 时，应触发告警并检查源库压力、网络带宽、消费能力和位点是否正确推进。

医疗数据采集必须遵守最小必要原则与脱敏要求。采集患者身份信息、诊断、检验结果时，应在采集端或入湖前完成字段级权限控制与脱敏标记，原始敏感数据不得随意落盘到非受控目录。

综合来看，采集方式的选择没有唯一答案：历史数据用批量，变化数据用增量或 CDC，设备数据用实时流，跨网数据用文件摆渡，外部服务用 API。真正决定采集工程质量的是调度是否可依赖、断点是否可恢复、失败是否可重试、异常是否可观测。把这四件事做扎实，多源医疗数据才能稳定、完整、可追溯地进入后续治理环节。