zsjz-ai/.qoder/repowiki/zh/content/AI 数据清洗/清洗结果与统计.md

清洗结果与统计

本文引用的文件

• AiRowCleanResult.java
• AiRejectRecord.java
• ColumnError.java
• AiRejectReason.java
• AiRecordCleaner.java
• AiRejectCollector.java
• AiDuckDb.java
• AiCleanService.java
• ai_clean.sql
• ai_clean_row_stats.sql
• cleanProgress.vue
• CleanLogModal.vue

目录

1. 引言
2. 模型结构与用途
3. 成功行、拒绝行、列级错误与拒绝原因的关系
4. 清洗结果持久化流程
5. 行数统计与 ai_clean_row_stats.sql 设计
6. 查看清洗结果与导出示例
7. 数据保留策略、清理时机与性能考虑
8. 常见问题排查
9. 结论

引言

AI 数据清洗在“智能清洗”链路中，把用户上传的文件按模板或 AI 推断规则逐行清洗：必填列出错会整行拒绝，非必填列出错只清空该单元格并继续入库。系统需要同时回答三件事：

• 哪些行进入了目标表；
• 哪些行没有进入，以及为什么；
• 哪些列、哪类错误最多，是否值得人工复核。

围绕这些目标，代码定义了 AiRowCleanResult、ColumnError、AiRejectReason、AiRejectRecord 四个核心概念，并在 DuckDB 案件库中落盘失败明细，在 ai_clean_job 和 ai_clean_batch 中聚合统计。

模型结构与用途

AiRowCleanResult：一行清洗的内存结果

字段	含义	说明
values	列名到清洗后值的映射	被拒绝的行也保留可用的部分值，用于排障
errors	本行所有列的错误列表	不只记录第一个阻塞错误
rejected	是否整行拒绝	存在任意 blocking 错误时为真

它还提供两个诊断方法：

• firstBlocking()：返回第一个导致整行拒绝的错误列；没有阻塞错误时为空。
• hasOptionalError()：是否存在非阻塞错误；这类错误不会阻止入库，但会把对应单元格置空。

这个模型是“单行清洗”的边界：它不关心数据库，也不直接写日志，只表达“这一行洗完后是什么状态”。

章节来源

• AiRowCleanResult.java:1-47

ColumnError：单列错误

字段	含义	说明
colIndex	原文件中的列序号	0 基
columnName	动态表物理列名	例如 c0、c1
reason	失败原因枚举	来自 AiRejectReason
valuePreview	原始值预览	超长截断，不做脱敏
blocking	是否导致整行拒绝	由原因枚举决定

ColumnError 是比 AiRejectRecord 更细的粒度：一个拒绝行可能包含多个列错误，但只有第一个 blocking 列会被选为“拒绝原因列”。

章节来源

• ColumnError.java:1-21

AiRejectReason：拒绝原因分类

枚举值	含义	是否阻塞	典型场景
REQUIRED_EMPTY	必填列为空	是	身份证号、金额等必填项缺失
REQUIRED_TYPE_INVALID	必填列类型解析失败	是	日期、数字、身份证、电话、卡号格式不对
REQUIRED_FUNC_ERROR	必填列清洗函数执行异常	是	自定义转换函数抛错
REQUIRED_LENGTH_OVERFLOW	必填列超长或精度溢出	是	数值超出目标长度/精度
OPTIONAL_TYPE_INVALID	非必填列类型解析失败	否	可选字段格式错误，单元格置空
OPTIONAL_FUNC_ERROR	非必填列清洗函数执行异常	否	可选字段转换异常
OPTIONAL_LENGTH_OVERFLOW	非必填列超长或精度溢出	否	可选字段超出长度/精度
WRITER_ERROR	写入数据库失败	是	DuckDB 批量写入异常

关键语义：

• 以 REQUIRED_ 开头的错误一律阻塞整行。
• WRITER_ERROR 也阻塞整行，因为它表示“这行没能成功落库”。
• label() 提供界面文案，如“必填列为空”“非必填列类型不合法（已置空）”。

章节来源

• AiRejectReason.java:1-49

AiRejectRecord：持久化的拒绝行

字段	含义
jobId	AI 清洗任务 id
caseId	案件 id
fileId	文件根 id
sheetId	sheet id
blockNo	表块序号
rowNo	原文件物理行号
blockingColIndex	第一个阻塞列的文件列序号
blockingColName	第一个阻塞列的物理列名
blockingColNameCn	第一个阻塞列的中文表头
reason	拒绝原因枚举名称
errorsJson	该行全部列错误的 JSON 序列化

一条 AiRejectRecord 对应一个被拒绝的数据行。它的价值在于：既能回答“这一行为什么没进去”，也能通过 errorsJson 还原整行的列级错误，从而支持“按列聚合看哪一列错得最多”。

章节来源

• AiRejectRecord.java:1-35

  classDiagram
  class AiRowCleanResult {
      +values
      +errors
      +rejected
      +firstBlocking()
      +hasOptionalError()
  }
  
  class ColumnError {
      +colIndex
      +columnName
      +reason
      +valuePreview
      +blocking
  }
  
  class AiRejectReason {
      +REQUIRED_EMPTY
      +REQUIRED_TYPE_INVALID
      +REQUIRED_FUNC_ERROR
      +REQUIRED_LENGTH_OVERFLOW
      +OPTIONAL_TYPE_INVALID
      +OPTIONAL_FUNC_ERROR
      +OPTIONAL_LENGTH_OVERFLOW
      +WRITER_ERROR
      +blocking()
      +label()
  }
  
  class AiRejectRecord {
      +jobId
      +caseId
      +fileId
      +sheetId
      +blockNo
      +rowNo
      +blockingColIndex
      +blockingColName
      +blockingColNameCn
      +reason
      +errorsJson
  }
  
  AiRowCleanResult --> ColumnError : "包含"
  ColumnError --> AiRejectReason : "使用"
  AiRejectRecord --> AiRejectReason : "reason 引用"
  

图表来源

• AiRowCleanResult.java:1-47
• ColumnError.java:1-21
• AiRejectReason.java:1-49
• AiRejectRecord.java:1-35

成功行、拒绝行、列级错误与拒绝原因的关系

清洗过程不是简单的“成功/失败二选一”，而是三层关系：

1. 列级错误：每一列都可能产生 ColumnError，包括空值、类型不合法、函数异常、超长溢出。
2. 行级判定：如果任一列错误是 blocking，则 AiRowCleanResult.rejected = true；否则行可以入库。
3. 拒绝原因：拒绝行取第一个 blocking 列的 AiRejectReason 作为 AiRejectRecord.reason。

因此：

• “成功行”指 rejected = false 的行，即使它们仍有非阻塞错误。
• “拒绝行”指 rejected = true 的行，最终会写入 ai_clean_reject。
• “列级错误”是诊断细节，用于解释“哪个列出了问题、为什么”。

• “拒绝原因分类”是用户可理解的标签，用于统计“必填列为空有多少、类型不合法有多少、写入失败有多少”。

  flowchart TD
  Start["读取一行原始数据"] --> CleanColumns["逐列清洗<br/>生成 ColumnError"]
  CleanColumns --> HasBlocking{"是否存在阻塞错误？"}
  HasBlocking -->|"是"| Reject["标记行拒绝<br/>写入 ai_clean_reject"]
  HasBlocking -->|"否"| Insert["写入目标表"]
  Reject --> OptionalErrors{"是否有非阻塞错误？"}
  OptionalErrors -->|"有"| Warn["计入 warned 数量"]
  OptionalErrors -->|"无"| End
  Insert --> End["完成"]
  

图表来源

• AiRecordCleaner.java:1-200
• AiRejectCollector.java:1-222

章节来源

• AiRecordCleaner.java:1-200
• AiRejectCollector.java:1-222

清洗结果持久化流程

清洗结果并不只保存在一张表里，而是分两层：

层级	存储位置	内容	用途
行级明细	DuckDB 案件库的 ai_clean_reject	每个被拒绝行一条记录	查“某行为什么没进”、按列统计
任务级统计	PostgreSQL 元数据表的 ai_clean_job、ai_clean_batch	总行数、成功数、拒绝数、警告数、未处理数	进度页、批次汇总、治理树

逐行清洗与拒绝收集

AiRecordCleaner.cleanRowDetailed 负责逐列清洗：

1. 对空值判断是否必填。
2. 执行函数链，捕获异常。
3. 若函数链把值清空，必填列按空值处理。
4. 对非字典类型进行类型规整。
5. 收集所有列错误。
6. 根据是否存在 blocking 错误决定整行拒绝。

AiRejectCollector 负责把拒绝行转成 AiRejectRecord，并按原因、列、非必填错误分别聚合计数。它还会采样前若干条拒绝日志，避免十万行文件中大量拒绝记录刷爆应用日志。

写入 ai_clean_reject

拒绝记录采用批缓冲写入：

• 默认每 500 条刷新一次。
• 写入目标为 DuckDB 案件库中的 ai_clean_reject 表。

• 每条记录包含 job、file、sheet、block、rowNo、首个阻塞列、原因和全部列错误 JSON。

  sequenceDiagram
  participant Cleaner as "AiRecordCleaner"
  participant Collector as "AiRejectCollector"
  participant DB as "AiDuckDb"
  participant Table as "ai_clean_reject"
  
  Cleaner->>Collector: record(rowNo, result)
  alt 行被拒绝
      Collector->>Collector: 更新 rejected / reasons / columns
      Collector->>DB: insertRejects(records)
      DB->>Table: INSERT 批量写入
  else 行成功但有非阻塞错误
      Collector->>Collector: 更新 warned / optionalCounts
  end
  

图表来源

• AiRecordCleaner.java:1-200
• AiRejectCollector.java:1-222
• AiDuckDb.java:350-360

章节来源

• AiRecordCleaner.java:1-200
• AiRejectCollector.java:1-222
• AiDuckDb.java:332-360

行数统计与 ai_clean_row_stats.sql 设计

任务级统计字段

ai_clean_job 中维护以下行级统计：

字段	含义
row_total	输入总行数
row_loaded	成功入库行数
row_rejected	因必填为空、类型错误、函数异常、超长或写入失败而整行拒绝的行数
row_warned	已入库但包含非必填错误的行数
row_unprocessed	失败任务中未处理到的行数

批次级统计字段

ai_clean_batch 中维护以下批次维度统计：

字段	含义
loaded_rows	出口 B 实际入库行数
rejected_rows	出口 B 逐行拒绝行数
unprocessed_rows	待人工、跳过或失败任务剩余行数
self_built_count	无模板自建表的 sheet 数

ai_clean_row_stats.sql 的作用

该脚本不重新建表，而是为已存在的 ai_clean_job 和 ai_clean_batch 增加统计列：

• 给 ai_clean_job 增加 row_rejected、row_warned、row_unprocessed。
• 给 ai_clean_batch 增加 loaded_rows、rejected_rows、unprocessed_rows、self_built_count。
• 新增列带默认值，语句幂等，可重复执行。

它与 ai_clean.sql 的关系是：

• ai_clean.sql：定义 AI 清洗元数据主表结构，包括模板、字段、规则、任务、批次、同义词库。

• ai_clean_row_stats.sql：在已有结构上扩展行级统计列，便于后续按任务/批次统计。

  graph TB
  Job["ai_clean_job<br/>row_total / row_loaded / row_rejected / row_warned / row_unprocessed"]
  Batch["ai_clean_batch<br/>loaded_rows / rejected_rows / unprocessed_rows / self_built_count"]
  Reject["ai_clean_reject<br/>拒绝行明细"]
  Template["ai_clean_template"]
  Field["ai_clean_field"]
  Rule["ai_clean_rule"]
  
  Template --> Job
  Field --> Job
  Rule --> Job
  Job --> Batch
  Job --> Reject
  

图表来源

• ai_clean.sql:1-214
• ai_clean_row_stats.sql:1-21

章节来源

• ai_clean.sql:1-214
• ai_clean_row_stats.sql:1-21

查看清洗结果与导出示例

查看清洗结果

前端“数据清洗进度”页面在任务完成后展示：

• 导入成功条数；
• 导入失败条数；
• 未清洗表数量；
• 成功率；
• 任务耗时；
• 处理日志。

点击“查看结果”会进入 AI 动态表的数据分页接口。后端 AiCleanService.dataPage 会：

1. 根据模板加载字段定义；
2. 用物理列名 c0..cN 查询 DuckDB；
3. 返回与既有治理树表格对齐的结构，包含 head、pages.records、pages.total 等；
4. 前端复用通用表格组件，无需为 AI 表单独实现翻页逻辑。

章节来源

• cleanProgress.vue:1-200
• AiCleanService.java:1000-1152

查看清洗日志

“查看清洗日志”打开 CleanLogModal，基于 governApi.cleanErrorLogList 查询 CleanErrorLog 相关日志。它支持：

• 按文件范围筛选；
• 按 sheet 范围筛选；
• 排序；
• 分页；
• 查看详情。

注意：这里主要展示的是“清洗日志”，不是 ai_clean_reject 的全部拒绝明细。真正的拒绝明细需要通过 SQL 查询 DuckDB 案件库中的 ai_clean_reject 表。

章节来源

• CleanLogModal.vue:1-200

导出成功数据

成功数据位于 AI 生成的动态表中，表名通常以 AI_ 前缀开头，例如“AI_银行交易流水”。可通过以下方式导出：

1. 在“数据治理”或“数据清洗结果”页找到对应 AI 表。
2. 确认当前案件已打开，因为动态表走案件路由的 DuckDB。
3. 使用数据库客户端连接当前案件的 DuckDB 实例。
4. 执行分页查询，例如按页取出全部行。
5. 将结果导出为 CSV 或 Excel。

由于后端分页接口已经封装了 AiDuckDb.selectPage 和 AiDuckDb.count，也可以通过调用后端 /govern/treeTablePage 的 AI 分流接口获取结构化数据。

导出错误数据

拒绝数据位于 DuckDB 案件库的 ai_clean_reject 表中，关键字段包括：

• job_id：关联 ai_clean_job。
• file_id、sheet_id、block_no、row_no：定位原始文件中的行。
• blocking_col_index、blocking_col_name、blocking_col_name_cn：首个阻塞列。
• reason：拒绝原因枚举名称。
• errors_json：该行全部列错误的 JSON，包含列名、原因、原值预览和是否阻塞。

常用查询方向：

• 按 job 查看所有拒绝行；
• 按 reason 统计拒绝原因分布；
• 按 blocking 列统计哪一列错误最多；
• 按 file_id、sheet_id 定位具体文件和工作表。

章节来源

• AiDuckDb.java:332-472

数据保留策略、清理时机与性能考虑

数据保留策略

• 成功数据：保存在 AI 动态表中，表名由模板或文件/sheet 名称决定，统一加 AI_ 前缀。
• 拒绝数据：保存在 ai_clean_reject 表中，按 job/file/sheet/block/rowNo 定位。
• 任务元数据：保存在 ai_clean_job 和 ai_clean_batch 中，包含状态、耗时、token 用量、统计摘要等。
• 模板与规则：保存在 ai_clean_template、ai_clean_field、ai_clean_rule 中，属于长期资产。

当前代码中没有统一的定时清理策略删除历史拒绝记录；拒绝记录的生命周期通常与任务生命周期一致。

清理时机

撤销接口 AiCleanService.revert 会在按 job 撤销时：

1. 删除对应 AI 动态表中的数据；
2. 删除该 job 对应的 ai_clean_reject 记录；
3. 将 job 状态设为 SKIP；
4. 写入撤销原因。

这意味着：如果用户主动撤销清洗结果，成功数据和拒绝记录都会清理。

章节来源

• AiCleanService.java:1060-1120

性能考虑

1. 拒绝记录批写入

AiRejectCollector 每 500 条拒绝记录 flush 一次，避免频繁单条插入。

2. 成功数据流式写入

AiDuckDb.BatchWriter 每 2000 行 flush 一次，采用流式写入器，避免在回调中全量 collect。这是处理大表的关键设计。

3. 日志采样

拒绝日志默认只打前若干条，而不是每行都打。可通过配置控制：

• zsjz.ai-clean.reject-log-sample：抽样条数；
• zsjz.ai-clean.reject-log-all：是否每行拒绝都打日志。

4. detail JSON 有界

VerifyReport 的 previewRows 限制为前 5 行，detail JSON 最大长度 8000，防止 jobs 列表查询被大对象撑爆。

5. 值预览截断

拒绝记录中的 valuePreview 只做超长截断，不做脱敏，长度为固定上限，避免过长值污染统计和日志。

章节来源

• AiRejectCollector.java:1-222
• AiDuckDb.java:1-260
• AiCleanService.java:1120-1152

常见问题排查

问题一：结果为空但任务成功

可能原因：

1. 任务匹配到了既有模板（出口 A），但没有走 AI 动态表（出口 B）。
2. 文件没有数据行，或者数据行都在标题、合并区、表头区域之外。
3. 任务状态是 NEED_REVIEW，表示需要人工处理，而不是自动入库。
4. 前端看到的是“导入成功/失败”统计口径，而不是 AI 动态表数据。

排查步骤：

• 检查 ai_clean_job.status 和 exit_code。
• 检查是否命中既有模板。
• 检查 ai_clean_job.row_total 是否为 0。
• 如果是出口 A，不要期待 AI 动态表中有数据，应检查目标业务表。

章节来源

• AiCleanService.java:850-1000

问题二：拒绝记录缺失

可能原因：

1. 行确实成功入库，只是某些非必填列被置空，这时不会产生拒绝记录。
2. 任务尚未完成，拒绝记录还在缓冲区。
3. 已执行过撤销操作，ai_clean_reject 已被删除。
4. 查询条件过滤掉了目标 job 或 sheet。

排查步骤：

• 检查任务状态是否已完成。
• 检查 ai_clean_job.row_rejected。
• 用 job_id 精确查询 ai_clean_reject。
• 检查是否调用了 revert。

章节来源

• AiRejectCollector.java:100-222
• AiDuckDb.java:451-472

问题三：错误列信息不准确

可能原因：

1. 用户关注的是“第一个阻塞列”，但实际有多个列错误。
2. 非必填列错误不会导致拒绝，但会影响 optionalCounts。
3. blockingColNameCn 可能为空，此时显示回退到物理列名。
4. 原值预览被截断，看不出完整脏值。

排查建议：

• 优先看 blocking_col_index 和 blocking_col_name。
• 再读 errors_json，查看该行所有列错误。
• 结合 ai_clean_field.field_name_cn 还原中文列名。
• 对于 WRITER_ERROR，检查 errors_json 中的消息，而不是依赖列名。

章节来源

• AiRejectRecord.java:1-35
• AiRejectCollector.java:150-222

问题四：统计数据延迟

可能原因：

1. 任务仍在运行，拒绝记录还在 AiRejectCollector.pending 缓冲区。
2. 前端轮询间隔较长。
3. 融合模式下 SSE 推送中断，前端无法实时感知进度。
4. 任务状态停留在中间阶段，尚未调用 finish。

排查建议：

• 直接查 ai_clean_job 最新状态。
• 检查 SSE 连接状态。
• 等待任务完成后再刷新。
• 如果进度中断，可在“数据治理”查看已入库结果，或改用手动清洗。

章节来源

• cleanProgress.vue:1-200
• AiRejectCollector.java:1-100

问题五：导出文件过大

可能原因：

1. AI 动态表数据量大，一次性导出整表。
2. errors_json 中包含原值预览，且未做脱敏。
3. 拒绝记录过多，尤其是超大文件或高错误率文件。

优化建议：

• 分页导出，不要一次性拉取全量。
• 只导出必要列，避免导出全部 errors_json。
• 对敏感值在前端或导出层做脱敏。
• 对历史拒绝记录制定归档或清理策略。
• 使用数据库侧过滤条件缩小范围，例如按 job、file、reason。

章节来源

• AiDuckDb.java:1-260
• AiRejectCollector.java:150-222

结论

AI 清洗结果的设计围绕“行级可诊断、列级可统计、任务级可观测”展开：

• AiRowCleanResult 表达单行清洗结果；
• ColumnError 表达单列错误；
• AiRejectReason 区分必填与非必填、类型/函数/长度/写入失败；
• AiRejectRecord 持久化拒绝行，支撑排障和聚合统计；
• ai_clean_job 和 ai_clean_batch 提供任务与批次维度的行数统计；
• ai_clean_row_stats.sql 负责扩展统计列，使现有表能承载新的统计需求；
• 前端通过进度页、日志弹窗和数据分页接口查看结果，真实拒绝明细通过 DuckDB 的 ai_clean_reject 表分析。

在大规模数据下，系统通过批写入、流式写入、日志采样和 JSON 有界控制性能；在可维护性上，通过撤销接口、job 唯一键和统计摘要保证结果可追溯、可清理、可对比。