批量清洗任务

本文引用的文件

AiCleanBatch.java
AiCleanJob.java
AiRecordCleaner.java
AiDuckDb.java
AiRowReader.java
SheetProbe.java
AiCleanService.java
AiSmartCleanService.java
AiCleanWatchdog.java
AiCleanStatusEnum.java
AiCleanStageEnum.java
ai_clean.sql
ai-clean-parallel.md

引言

本文聚焦 AI 数据清洗中的批量清洗任务，围绕 AiCleanBatch、AiCleanJob、AiRecordCleaner、AiDuckDb、AiRowReader、SheetProbe 等核心组件，说明用户上传 CSV/Excel 后，系统如何解析 Sheet、读取行、匹配清洗模板、执行清洗规则、写入 DuckDB 临时表或目标表，并生成清洗结果与错误记录。同时给出批量任务从创建到完成的全生命周期、状态枚举、失败重试机制、超时处理、并发限制，以及常见故障的排查建议。

项目结构

AI 批量清洗相关代码主要位于后端 ai-server 模块中：

实体层：AiCleanBatch（批次）、AiCleanJob（任务）
执行层：AiRecordCleaner（逐行清洗）、AiDuckDb（DuckDB 读写）
探测与读取层：SheetProbe（结构探测）、AiRowReader（源文件流式读取）
编排与服务层：AiSmartCleanService（融合编排）、AiCleanService（S1→S7 主流程）、AiCleanWatchdog（看门狗）

枚举与 SQL：AiCleanStatusEnum、AiCleanStageEnum、ai_clean.sql

graph TB
Frontend["前端上传页面"] --> Smart["AiSmartCleanService<br/>智能清洗编排"]
Smart --> CleanSvc["AiCleanService<br/>S1→S7 主流程"]
CleanSvc --> Probe["SheetProbe<br/>结构探测"]
CleanSvc --> RowReader["AiRowReader<br/>源文件流式读取"]
CleanSvc --> Cleaner["AiRecordCleaner<br/>逐行清洗"]
CleanSvc --> Duck["AiDuckDb<br/>DuckDB 读写通道"]
Duck --> Db[("案件 DuckDB<br/>业务表 / ai_t{n} / raw_{sheetId}")]
Smart --> Watch["AiCleanWatchdog<br/>批次看门狗"]

图示来源

AiSmartCleanService.java:1-200
AiCleanService.java:1-200
SheetProbe.java:1-193
AiRowReader.java:1-121
AiRecordCleaner.java:1-200
AiDuckDb.java:1-200
AiCleanWatchdog.java:1-95

章节来源

AiCleanBatch.java:1-99
AiCleanJob.java:1-165
ai_clean.sql:1-200

核心组件

本节按职责说明六个关键组件的职责、输入输出和与其他组件的关系。

组件	职责	主要输入	主要输出	关键行为
`AiCleanBatch`	表示一次“智能清洗”批次，承载批次级状态机与幂等锚点	caseId、batchId、文件树	阶段、计数、待人工明细、最后错误	使用累加列避免频繁 count；作为批次进度与终态依据
`AiCleanJob`	表示一个 sheet/block 的任务单元，是幂等游标、独立状态线与成本埋点	caseId、fileId、sheetId、blockNo、ver	状态、出口、命中途径、模型调用量、行数统计	唯一键 `(caseId, fileId, sheetId, blockNo, ver)` 保证幂等
`SheetProbe`	S1 结构探测：扫描源文件前若干行，提取表块、合并区、列统计、指纹	caseId/fileId/sheetId、文件名、sheetName、源文件	`SheetEvidence`	CSV 用 `AiRowReader` 读前 200 行；大文件或 POI 失败回退
`AiRowReader`	从源文件流式读取「列序号 → 值」的行，用于预览、校验与兜底读取	文件、sheetName、headerRow	行集合或回调	支持 XLS/XLSX/CSV；过滤空行；流式不攒全量
`AiRecordCleaner`	把原始行按编译后的列计划做类型规整、函数链执行、必填校验	原始行 Map、`CompiledColumn` 列表	清洗后值、列级错误、是否整行拒绝	非阻塞错误置空单元格；阻塞错误导致整行拒绝
`AiDuckDb`	在案件 DuckDB 上建表、批量写入、撤销、失败明细、分页查询	表名、列名、行数据、jobId	写入行数、失败明细页、临时表清理结果	批大小 2000；失败返回 -1；fail-open 写失败明细

章节来源

AiCleanBatch.java:1-99
AiCleanJob.java:1-165
SheetProbe.java:1-193
AiRowReader.java:1-121
AiRecordCleaner.java:1-200
AiDuckDb.java:1-200

架构总览

批量清洗不是“一个接口直接洗一行”，而是由批次编排 + 任务调度 + 单任务主流程 + 数据读写通道组成的流水线。

sequenceDiagram
    participant FE as "前端"
    participant Smart as "AiSmartCleanService"
    participant Batch as "AiCleanBatch"
    participant Clean as "AiCleanService"
    participant Probe as "SheetProbe"
    participant Reader as "AiRowReader"
    participant Cleaner as "AiRecordCleaner"
    participant Duck as "AiDuckDb"
    participant DB as "DuckDB"

    FE->>Smart: 提交智能清洗(文件树)
    Smart->>Batch: upsertBatch(批次数组初始化)
    Smart-->>FE: 返回 batchId + JUDGING
    Smart->>Clean: 对未匹配 sheet 提交 job
    Clean->>Probe: 结构探测
    Probe-->>Clean: SheetEvidence
    Clean->>Clean: 模板匹配 + 置信度校验
    alt 命中既有模板
        Clean->>DB: 走既有 doClean(出口 A)
    else 无合适模板
        Clean->>Duck: 自建表/写入动态表(出口 B)
        Clean->>DB: 写入 ai_t{n}
    end
    Clean->>Duck: 收集拒绝行(ai_clean_reject)
    Smart->>Batch: 更新阶段/计数
    Smart-->>FE: SSE 推送进度/终态

图示来源

AiSmartCleanService.java:1-200
AiCleanService.java:1-200
SheetProbe.java:1-193
AiDuckDb.java:1-200
ai-clean-parallel.md:1-93

端到端流程

用户视角

用户在上传页选择 CSV/Excel，点击“智能”。
后端立即返回 batchId，批次进入“AI 判定中”。
后端并行探测 Sheet 结构、匹配模板。
命中既有模板的 Sheet 并入既有清洗链路；没有合适模板的 Sheet 由 AI 自建模板并写入 DuckDB 动态表。
前端通过 SSE 和轮询查看进度、待人工明细、最终结果。

后端视角

AiSmartCleanService.submit 接收文件树，构造批次，复用已有批次，启动虚拟线程异步运行。
对未匹配模板的 sheet，AiCleanService.submitUnmatched 或 prepareJob 创建 AiCleanJob。
AiCleanService 执行 S1→S7：结构探测、结构归一、模板匹配、规则编译、抽样校验、双出口入库。

成功时更新批次计数；失败时记录 lastError，由看门狗或重跑恢复。

flowchart TD
Start["用户上传 CSV/Excel"] --> Submit["后端创建智能清洗批次"]
Submit --> Judge["并行判定：结构探测 + 模板匹配"]
Judge --> Matched{"是否命中既有模板？"}
Matched -- "是" --> Existing["既有链路清洗<br/>数据落业务表"]
Matched -- "否" --> SelfBuild["AI 自建模板 + 规则"]
SelfBuild --> Load["写入 DuckDB 动态表 ai_t{n}"]
Existing --> Collect["收集拒绝行/统计行数"]
Load --> Collect
Collect --> Update["更新 AiCleanBatch 计数与阶段"]
Update --> Done{"是否全部完成？"}
Done -- "是" --> Final["DONE/PARTIAL/SUCCESS/FAIL"]
Done -- "否" --> Continue["继续处理剩余 Job"]

图示来源

AiSmartCleanService.java:1-200
AiCleanService.java:1-200
ai-clean-parallel.md:1-93

章节来源

AiSmartCleanService.java:1-200
AiCleanService.java:1-200
ai_clean.sql:1-200

详细组件分析

`AiCleanBatch`：批次状态机与幂等锚点

AiCleanBatch 是批量清洗任务的“容器”。它记录：

批次标识 batchId
当前阶段 stage
文件数、Sheet 总数、已匹配数、AI 判定数、既有命中数、AI 入库数
实际入库行数、拒绝行数、未处理行数
待人工明细 JSON、最后错误、创建/更新时间

设计要点：

计数采用累加列而非每次 count(*)，避免前端高频轮询给主库带来压力。
唯一键 (caseId, batchId) 保证重复提交复用同一批次。

阶段推进顺序为：JUDGING → CLEANING_EXISTING → LOADING_AI → DONE/PARTIAL/FAILED。

stateDiagram-v2
[*] --> JUDGING : "提交智能清洗"
JUDGING --> CLEANING_EXISTING : "判定完成，既有模板可复用"
CLEANING_EXISTING --> LOADING_AI : "既有清洗完成，开始 AI 入库"
LOADING_AI --> DONE : "全部成功"
LOADING_AI --> PARTIAL : "部分完成"
LOADING_AI --> FAILED : "批次异常"
JUDGING --> FAILED : "判定阶段异常"
CLEANING_EXISTING --> FAILED : "既有链路异常"

图示来源

AiCleanBatch.java:1-99
AiCleanStageEnum.java:1-52
ai_clean.sql:1-200

章节来源

AiCleanBatch.java:1-99
AiCleanStageEnum.java:1-52
ai_clean.sql:1-200

`AiCleanJob`：幂等任务单元与状态线

AiCleanJob 是单个 Sheet 或表块的执行单元，关键字段包括：

caseId/fileId/sheetId/blockNo/ver：构成幂等唯一键
status：任务状态，独立于既有文件状态
exitCode/matchedBy/matchTier：出口与命中途径
refTemplateId/aiTemplateId：出口 A/B 对应模板
llmCalls/tokenIn/tokenOut/costMs：模型调用成本埋点
rowTotal/rowLoaded/rowRejected/rowWarned/rowUnprocessed：行级统计

设计要点：

唯一键 (caseId, file_id, sheet_id, block_no, ver) 同时解决轮询器捞取、重跑 UPSERT、并发重复提交。
状态线独立于既有 FileSheetStateEnum，防止 AI 辅助链路影响主展示。

终态包含 SUCCESS/NEED_REVIEW/FAIL/SKIP/SUPERSEDED。

classDiagram
class AiCleanJob {
    +Long id
    +Long caseId
    +Long fileId
    +Long sheetId
    +Integer blockNo
    +Integer ver
    +Long batchId
    +String fileName
    +String sheetName
    +String status
    +String exitCode
    +String matchedBy
    +String matchTier
    +Integer refTemplateId
    +Long aiTemplateId
    +Integer rowTotal
    +Integer rowLoaded
    +Integer rowRejected
    +Integer rowWarned
    +Integer rowUnprocessed
    +String failReason
}

图示来源

AiCleanJob.java:1-165
AiCleanStatusEnum.java:1-76
ai_clean.sql:1-200

章节来源

AiCleanJob.java:1-165
AiCleanStatusEnum.java:1-76
ai_clean.sql:1-200

`SheetProbe`：结构探测

SheetProbe 负责 S1 阶段的结构探测：

扫描最多 200 行，识别表块、合并区域、列统计、表头指纹。
CSV 不走 POI，改用 AiRowReader 读前 200 行，保持与入库一致的取值口径。
超过 80MB 的文件跳过结构探测，交由无结构路径处理。
输出 SheetEvidence，供后续模板匹配与置信度校验使用。

关键点：

必须回源文件是因为上传预览表中合并单元格除左上格外已丢失，无法还原真实结构。

探测只读不写，不会污染业务数据。

flowchart TD
Input["源文件 + sheetName"] --> CsvCheck{"是否为 CSV？"}
CsvCheck -- "是" --> UseReader["使用 AiRowReader 读前 200 行"]
CsvCheck -- "否" --> SizeCheck{"文件大小 ≤ 80MB？"}
SizeCheck -- "否" --> Skip["跳过结构探测"]
SizeCheck -- "是" --> PoiRead["POI 读取表头 + 合并区"]
UseReader --> Build["构建 BlockHint/ColStat/HeaderFingerprint"]
PoiRead --> Build
Build --> Evidence["返回 SheetEvidence"]

图示来源

SheetProbe.java:1-193
AiRowReader.java:1-121

章节来源

SheetProbe.java:1-193
AiRowReader.java:1-121

`AiRowReader`：源文件流式读取

AiRowReader 提供两种读取方式：

read：收集型读取，用于样本/校验场景。
forEachRow：流式读取，用于正式入库，避免内存堆积。

关键点：

支持 XLS/XLSX/CSV，自动识别后缀。
跳过表头行及上方内容，跳过整行为空的行。
转换器使用 Str2NullConverter 和 DateNumberConverter，与预览表一致。

失败返回 false，不能当成“读完”；调用方需据此放弃本轮入库。

flowchart TD
Start["forEachRow(file, sheetName, headerRow)"] --> CheckFile{"文件存在？"}
CheckFile -- "否" --> ReturnFalse["返回 false"]
CheckFile -- "是" --> DetectType["根据后缀判断 XLS/XLSX/CSV"]
DetectType --> Fesod["Fesod 流式读取"]
Fesod --> Filter["跳过表头行 + 空行"]
Filter --> Callback["回调 handler.accept(row, lineNo)"]
Callback --> End["返回 true"]

图示来源

AiRowReader.java:1-121

章节来源

AiRowReader.java:1-121

`AiRecordCleaner`：逐行清洗

AiRecordCleaner 是清洗引擎的核心：

遍历每个 CompiledColumn，执行去特殊字符、函数链、类型规整。
区分必填与非必填列：必填列出错整行拒绝；非必填列出错仅该单元格置空。
方向字典翻译：借贷标志、通话方向等字段通过反向字典归一为标准词。
输出 AiRowCleanResult，包含清洗值、列错误、是否整行拒绝。

关键点：

函数链异常被捕获为列错误，而不是静默吞掉。
预览长度截断 80 字符，不做脱敏。

方向字典缓存上限 256，超限整体清空重建。

flowchart TD
Start["cleanRowDetailed(rowData, columns)"] --> Init["初始化 values + errors"]
Init --> Loop["遍历 CompiledColumn"]
Loop --> Raw["取原始值"]
Raw --> Blank{"值为空？"}
Blank -- "是且必填" --> RequiredEmpty["标记 REQUIRED_EMPTY"]
Blank -- "是且非必填" --> OptionalEmpty["置空单元格"]
Blank -- "否" --> PreClean["预清洗：去控制字符/半角转换"]
PreClean --> FunChain["执行函数链"]
FunChain --> FunError{"函数异常？"}
FunError -- "是" --> FunReject["标记函数错误"]
FunError -- "否" --> Coerce["类型规整"]
Coerce --> CoerceOk{"规整成功？"}
CoerceOk -- "否" --> TypeReject["标记类型/长度错误"]
CoerceOk -- "是" --> WriteValue["写入清洗值"]
RequiredEmpty --> AnyBlocking{"是否有阻塞错误？"}
OptionalEmpty --> AnyBlocking
FunReject --> AnyBlocking
TypeReject --> AnyBlocking
WriteValue --> AnyBlocking
AnyBlocking -- "有" --> Reject["整行拒绝"]
AnyBlocking -- "无" --> Success["行清洗成功"]

图示来源

AiRecordCleaner.java:1-200
AiRecordCleaner.java:201-312

章节来源

AiRecordCleaner.java:1-200
AiRecordCleaner.java:201-312

`AiDuckDb`：DuckDB 读写通道

AiDuckDb 是 AI 侧访问案件 DuckDB 的统一通道：

建表：createAiTable 按字段定义生成 DDL。
批量写入：insertBatch 使用 BatchWriter，批大小 2000。
流式写入：openWriter 返回 BatchWriter，适合大表全量入库。
撤销：deleteByJob 按 job 精确删除，支持重跑。
失败明细：insertRejects 写入 ai_clean_reject，fail-open。
预览表读取：rawSheet、forEachRawRow 优先从 raw_{sheetId} 读取。
临时表清理：dropTablesByPrefix 清理上传阶段临时表。

关键点：

所有方法需在案件上下文作用域内调用，否则路由不到库。
写入失败返回 -1，不能当成成功。

失败明细写失败不影响主清洗流程。

classDiagram
class AiDuckDb {
    +runStatements(sqls) String
    +createAiTable(table, fields) String
    +insertBatch(table, columns, rows) long
    +openWriter(table, columns) BatchWriter
    +deleteByJob(table, jobId) long
    +deleteByFileSheet(table, fileId, sheetId) long
    +insertRejects(records) void
    +rejectPage(jobId, reason, page, size) Map
    +rawSheet(sheetId, expectedRows) RawSheet
    +forEachRawRow(raw, headerRow, handler) void
    +dropTablesByPrefix(prefix) int
}
class BatchWriter {
    +add(row) void
    +finish() long
    +close() void
}
AiDuckDb --> BatchWriter : "openWriter 返回"

图示来源

AiDuckDb.java:1-200
AiDuckDb.java:201-557
AiDuckDb.java:558-787

章节来源

AiDuckDb.java:1-200
AiDuckDb.java:201-557
AiDuckDb.java:558-787

依赖关系分析

组件之间的依赖如下：

graph LR
    AiSmartCleanService --> AiCleanBatch
    AiSmartCleanService --> AiCleanService
    AiCleanService --> SheetProbe
    AiCleanService --> AiRowReader
    AiCleanService --> AiRecordCleaner
    AiCleanService --> AiDuckDb
    AiCleanService --> AiCleanJob
    AiCleanWatchdog --> AiCleanBatch

图示来源

AiSmartCleanService.java:1-200
AiCleanService.java:1-200
AiCleanWatchdog.java:1-95

章节来源

AiSmartCleanService.java:1-200
AiCleanService.java:1-200
AiCleanWatchdog.java:1-95

性能与并发特性

并发限制

AiSmartCleanService 使用虚拟线程池，并通过配置项 zsjz.ai-clean.max-concurrent 控制并发上限，默认值为 3。
每任务一个虚拟线程，真正并发由闸门控制，避免无限并发。

内存特性

AiRowReader.forEachRow 流式读取，不在内存攒全量列表。
AiDuckDb.BatchWriter 每 2000 行 flush 一次，堆占用恒定。
SheetProbe 只扫描前 200 行，大文件跳过结构探测。

超时与看门狗

AiCleanWatchdog 在进程重启时将非终态批次标为 FAILED，允许用户重新发起。
运行时心跳超时：非终态且 update_time 超过 zsjz.ai-clean.stale-minutes（默认 30 分钟）未推进，标为 FAILED。

模型调用成本

AiCleanJob.llmCalls 记录模型调用次数，直通命中应为 0。
AiSmartCleanService.batchTokenBudget 控制批次 token 预算，超限后剩余表转人工。

章节来源

AiSmartCleanService.java:1-200
AiCleanWatchdog.java:1-95
AiRecordCleaner.java:1-200
AiDuckDb.java:1-200

常见问题排查

批量任务卡住

现象：批次长期停留在 JUDGING/CLEANING_EXISTING/LOADING_AI，前端无进展。
可能原因：

虚拟线程被阻塞（模型流挂死、DB 连接挂起）。
进程重启后批次线程丢失，但批次仍为非终态。
排查建议：
检查 AiCleanWatchdog.failStale 是否将批次标为 FAILED。
查看 AiCleanBatch.lastError 获取最后错误信息。
确认 zsjz.ai-clean.stale-minutes 配置是否合理。

章节来源

AiCleanWatchdog.java:1-95
AiCleanBatch.java:1-99

大文件内存不足

现象：清洗大文件时 OOM 或 GC 频繁。
可能原因：

使用了非流式读取（如旧版 collect 全量）。
结构探测对超大文件加载整个 workbook。
排查建议：
确认使用 AiRowReader.forEachRow 而非 read。
确认 SheetProbe 对 >80MB 文件跳过结构探测。
确认 AiDuckDb.BatchWriter 批大小适中，避免单次 flush 过大。

章节来源

AiRowReader.java:1-121
SheetProbe.java:1-193
AiDuckDb.java:1-200

DuckDB 写入失败

现象：清洗显示成功但实际无数据，或动态表为空。
可能原因：

insertBatch 返回 -1，但调用方误判为成功。
批量写入中途失败，半截数据留在表中。
排查建议：
检查 AiDuckDb.insertBatch 返回值，-1 应视为失败。
使用 AiDuckDb.deleteByJob 按 job 清理半截数据。
检查 ai_clean_reject 表是否有拒绝记录。

章节来源

AiDuckDb.java:1-200
AiDuckDb.java:201-557

模板中途变更

现象：模板变更后，历史任务仍按旧模板清洗。
可能原因：

AiCleanJob.ver 版本未递增，旧 job 被复用。
模板唯一键 header_md5 未变化，命中缓存。
排查建议：
变更模板后重新触发清洗，确保 ver+1。
检查 AiCleanJob.refTemplateId/aiTemplateId 是否指向新模板。
使用 submitUnmatched 重新提交未匹配 job。

章节来源

AiCleanJob.java:1-165
AiCleanService.java:1-200

任务状态不一致

现象：AiCleanJob.status 与 AiCleanBatch.stage 不一致。
可能原因：

批次阶段推进与单任务状态推进不同步。
看门狗将批次标 FAILED，但单任务仍为中间态。
排查建议：
以 AiCleanBatch.stage 作为批次级终态判断。
单任务状态用于细粒度诊断，批次状态用于前端进度环。
检查 AiCleanWatchdog 是否将批次标为 FAILED。

章节来源

AiCleanStatusEnum.java:1-76
AiCleanStageEnum.java:1-52
AiCleanWatchdog.java:1-95

结论

AI 批量清洗任务通过 AiSmartCleanService 编排、AiCleanService 执行、SheetProbe 与 AiRowReader 读取、AiRecordCleaner 清洗、AiDuckDb 写入，形成一条完整的数据清洗流水线。AiCleanBatch 与 AiCleanJob 分别承担批次状态与任务状态，配合 AiCleanWatchdog 实现超时恢复。系统在并发、内存、失败语义、幂等方面做了细致设计，适用于大规模 CSV/Excel 数据的自动化清洗场景。

批量清洗任务.md 29 KB

История Исходник

批量清洗任务

目录

引言

项目结构

核心组件

架构总览

端到端流程

用户视角

后端视角

详细组件分析

`AiCleanBatch`：批次状态机与幂等锚点

`AiCleanJob`：幂等任务单元与状态线

`SheetProbe`：结构探测

`AiRowReader`：源文件流式读取

`AiRecordCleaner`：逐行清洗

`AiDuckDb`：DuckDB 读写通道

依赖关系分析

性能与并发特性

并发限制

内存特性

超时与看门狗

模型调用成本

常见问题排查

批量任务卡住

大文件内存不足

DuckDB 写入失败

模板中途变更

任务状态不一致

结论

批量清洗任务.md 29 KB История Исходник

批量清洗任务

目录

引言

项目结构

核心组件

架构总览

端到端流程

用户视角

后端视角

详细组件分析

AiCleanBatch：批次状态机与幂等锚点

AiCleanJob：幂等任务单元与状态线

SheetProbe：结构探测

AiRowReader：源文件流式读取

AiRecordCleaner：逐行清洗

AiDuckDb：DuckDB 读写通道

依赖关系分析

性能与并发特性

并发限制

内存特性

超时与看门狗

模型调用成本

常见问题排查

批量任务卡住

大文件内存不足

DuckDB 写入失败

模板中途变更

任务状态不一致

结论

批量清洗任务.md 29 KB

История Исходник

`AiCleanBatch`：批次状态机与幂等锚点

`AiCleanJob`：幂等任务单元与状态线

`SheetProbe`：结构探测

`AiRowReader`：源文件流式读取

`AiRecordCleaner`：逐行清洗

`AiDuckDb`：DuckDB 读写通道