4. 字段完整性校验：非空检查、默认值填充、字段缺失率统计、异常空值告警阈值设定

字段完整性是实时风控数据质量的基石。缺失字段会导致特征计算失败、模型输出偏差甚至规则引擎空指针异常。本章围绕四个核心动作展开：非空检查、默认值填充、缺失率统计与告警阈值设定，形成从检测到修复再到监控的闭环。

4.1 非空检查：实时拦截空值流入

非空检查是字段完整性校验的第一道防线。在实时风控流水线中，通常对以下关键字段强制非空：

• 实体标识：user_id、device_id、order_id
• 时间戳：event_time、request_time
• 金额/数值：transaction_amount、loan_amount
• 枚举状态：risk_level、decision_code

-- 实时风控非空检查示例（Flink SQL / ClickHouse）
SELECT
  user_id,
  event_time,
  transaction_amount,
  CASE
    WHEN user_id IS NULL THEN 'EMPTY_USER_ID'
    WHEN event_time IS NULL THEN 'EMPTY_EVENT_TIME'
    WHEN transaction_amount IS NULL THEN 'EMPTY_AMOUNT'
    ELSE 'PASS'
  END AS null_check_result
FROM raw_event_stream;

4.2 默认值填充：空值降级策略

对于非关键字段或业务允许缺失的字段，采用默认值填充策略，保证下游计算不中断。常见填充规则如下：

4.3 字段缺失率统计：量化数据质量

缺失率统计用于衡量字段的完整程度，是数据质量监控的核心指标。通常按时间窗口（如 5 分钟、1 小时）聚合计算：

-- 字段缺失率统计（按小时窗口）
SELECT
  toStartOfHour(event_time) AS window_start,
  field_name,
  COUNT(*) AS total_records,
  SUM(CASE WHEN field_value IS NULL THEN 1 ELSE 0 END) AS null_count,
  ROUND(
    SUM(CASE WHEN field_value IS NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2
  ) AS null_rate_pct
FROM field_quality_log
WHERE event_time >= now() - INTERVAL 24 HOUR
GROUP BY window_start, field_name
ORDER BY window_start DESC, null_rate_pct DESC;

缺失率统计结果通常写入监控看板，用于识别突发性数据质量问题。以下为典型缺失率分级：

4.4 异常空值告警阈值设定

告警阈值是缺失率监控的触发条件，设定过松会漏报，过紧则产生大量噪音。推荐采用动态基线 + 静态阈值组合策略：

• 静态阈值：固定百分比，如单字段缺失率超过 5% 触发告警。
• 动态基线：基于历史 7 天同一时间窗口的缺失率均值 ± 3σ 作为浮动阈值。
• 组合规则：当缺失率同时超过静态阈值和动态基线时，才触发告警，减少误报。

4.5 实时告警与闭环处理

当缺失率超过告警阈值时，系统应自动执行以下动作：

1. 日志记录：将异常字段、缺失率、时间窗口写入告警日志表。
2. 通知推送：通过钉钉/企微/邮件通知数据运维负责人。
3. 自动降级：对于严重缺失（如 > 10%）的字段，自动切换为备用数据源或默认值策略。
4. 根因分析：联动上游数据源健康检查，定位是采集端故障、传输丢包还是解析异常。

-- 告警日志表结构示例
CREATE TABLE alert_log (
  alert_id        UInt64,
  field_name      String,
  window_start    DateTime,
  null_rate_pct   Float32,
  threshold_pct   Float32,
  alert_level     String,  -- 'WARNING', 'CRITICAL'
  created_at      DateTime DEFAULT now(),
  resolved_at     Nullable(DateTime),
  status          String DEFAULT 'OPEN'  -- 'OPEN', 'ACKED', 'RESOLVED'
) ENGINE = MergeTree()
ORDER BY (alert_id, created_at);