合规数据上报（WA）子系统设计

WA 合规上报子系统以「CSV 元数据驱动的字段映射 + Redis 状态协调 + 文件后缀状态机 + Java/Shell 混合流水线 + NAS 双挂载交接」为核心设计，在多节点并发的约束下用「关键临界区加锁、可容忍竞态放行」的务实一致性策略，实现了 15 类监管数据集从 Kafka 实时流到加密 ZIP 经 VPN/FTP 报送的全自动链路，满足「至少每 5 分钟一包、每包不超限、数据可溯源」的合规约束。

1. 背景与合规要求

按照有关部门对互联网内容/服务提供者的监管要求，平台需将用户行为明细数据以离线文件（BCP 格式）+ 加密 ZIP + FTP 上传的方式主动报送，数据须可溯源到人、到地、到设备。

上报规范要点（决定本子系统全部设计的约束来源）：

• 每类业务一个数据集代码，每个数据集一份字段结构（BCP 列定义），列间以 TAB 分隔；
• 单个 BCP 文件中记录数不得超过 5000 条;
• 每个 ZIP 包内 BCP 文件数不超过 100 个，须附 GAB_ZIP_INDEX.xml 索引文件，描述本包包含的文件清单、行数与 BCP 数据结构；
• ZIP 上报原则：达到 5 分钟推送 或 100 个 BCP，满足其中任意一条件即打包推送；
• 每条记录须带数据产生地（GB/T 2260 六位行政区划码）、IP（数字型长整数）、经纬度（3 位整数 5 位小数）等溯源字段；
• 文件命名遵循「网络应用类型代码 – 时间戳 – 序号 – 数据集代码」规范，ZIP 需设解压口令；
• 上报节点、业务类型等通用信息固定写入索引（发起节点→目的节点，业务服务类型为「主动上报类」）。

本子系统覆盖 15 类数据集（14 个上报目录，视频评论与话题评论共用一个数据集）：

2. 总体架构

flowchart TB
    subgraph IN["数据源 - Kafka"]
        A1["埋点 Topic"]
        A2["Canal binlog Topic
用户/订单库
好友库等"]
    end

    subgraph STREAM["流处理层 - 每数据集一个独立流应用（15个wa* Pod）"]
        B1["过滤与校验
有效userId、事件类型、
表路由"]
        B2["字段映射
CSV配置驱动的
监管字段到业务字段
转换"]
        B3["富化
行政区划GB2260/IP长整型
/经纬度格式化/视频元数据
/支付单号/CDN前缀"]
        B4["追加写BCP文件
5000行滚动切分"]
    end

    subgraph TRIGGER["打包触发 - 双保险"]
        T1["条件触发
BCP文件数达到阈值6"]
        T2["定时触发
Redis key 5分钟过期事件
保证至少每5分钟一包"]
    end

    subgraph PIPE["打包流水线 - Java调用Shell"]
        C1["wa_bcp_format.sh
清洗NUL字符/空行/首列"]
        C2["Java补列
awk提取首列userId
批量查ES画像
生成补充列文件"]
        C3["wa_bcp_add.sh
cut裁列 + paste横向合并"]
        C4["生成 GAB_ZIP_INDEX.xml
文件清单+行数+数据结构"]
        C5["wa_zip.sh
带口令ZIP并移除源文件"]
    end

    subgraph UP["上传层 - 宿主机 crontab 每分钟"]
        D1["wa_ftp_upload.sh
VPN探活自愈 + FTP上传"]
        D2["strongSwan IPSec VPN"]
        D3["监管侧FTP"]
    end

    NAS[("共享 NAS
K8s PVC挂载 /data
宿主机NFS挂载/xxx")]
    R[("Redis
文件名/计数/自增序号
文件锁/zip抢占/TTL触发")]

    A1 --> B1
    A2 --> B1
    B1 --> B2 --> B3 --> B4 --> NAS
    B4 --> T1
    B4 -- setex 300s --> R
    R -- keyspace 过期事件 --> T2
    T1 --> C1
    T2 --> C1
    C1 --> C2 --> C3 --> C4 --> C5 --> NAS
    NAS --> D1 --> D2 --> D3
    B4 -.状态读写.-> R

责任域划分（理解本设计的关键）：

3. 元数据驱动：一个枚举 + 一张 CSV 决定一切

3.1 WaEnum：数据集注册中心

每个数据集在 WaEnum 中注册四元组：数据集代码、Redis key、落盘目录（/data/wa/<数据集代码>）、BCP 结构 CSV 路径。枚举构造时即完成两件事：

1. 解析 CSV 生成「BCP 字段 → 业务数据源字段」映射表；
2. 扫描映射表中的数据源类型（es.* / as.*），推导出该数据集需要补充的外部数据源清单。

新增一个数据集 = 加一个枚举项 + 一张 CSV + 一个流处理类，其余（目录、命名、补列、打包、上传）全部复用。

3.2 CSV 字段映射机制（声明式，非硬编码）

每个数据集一张 TSV 配置（如 user_register_item.csv），四列：

key（BCP字段名）   eng（英文名）   rmk（中文含义）   cv_key（业务来源，格式：库.表.字段）

映射执行算法（WaUtil.transToWaBcpData）：

1. 深拷贝该数据集的映射模板（每次转换基于干净模板，避免并发/复用污染）；
2. 把当前业务数据（binlog 行 / 埋点对象）组装成 数据源.字段名 → 值 的扁平 map（埋点对象经 BeanUtils.describe 转属性 map，binlog 行天然是列名 map）；
3. 遍历模板（而非业务数据）：模板 value 与业务 map 的 key 匹配则填入，匹配不上置空。

为什么遍历模板而不是业务数据：多个 BCP 字段可来自同一业务字段（如「用户 ID」与「注册账号」都取自用户表的 id 列）。以模板 value 做匹配 key，同一业务字段可被重复消费；同时保证输出列序 = CSV 配置列序（BCP 格式硬性要求），未匹配字段占位为空而非缺列。

4. 流处理层：原始数据 → 合规记录

4.1 拓扑模式

每个数据集一个独立流应用（@Profile("wa*")），拓扑为「无窗口旁路处理」：

stream(源 topic)
  → filter（表路由 / 事件类型 / 有效 userId 校验）
  → selectKey（统一为数据源标识，供映射函数识别）
  → foreach（字段映射 + 富化 + 追加写 BCP）

三类典型数据源接入：

4.2 富化逻辑（合规溯源字段的组装）

所有数据集都要补的公共字段：

• DATA_LAND（行政区划）：拿 userId 反查 ES 画像的省/市/区 → GB2260Util 在本地行政区划字典（2020 版 GB/T 2260 三级 JSON，约 19 万字符）中逐级匹配转 6 位码；对直辖市做「层级提升」特判（北京市→北京市→东城区，匹配时区名提升到市一级）；
• IP：ES 画像中的字符 IP 先正则校验合法性，再按监管特定的公式转数字型长整数；
• 经纬度：按监管特定规则统一格式化处理；
• ACTION_TIME：记录处理时刻的秒级时间戳。

各数据集的个性化富化举例：

• 注册：区分三个来源表——主表 INSERT 补「注册用户」类型；手机/账号表以 user_id 回填账号字段；实名表只取审核通过的记录，补「身份证 + 证件类型码」；头像拼 CDN 域名前缀；注册时间格式化；
• 视频播放：按 video_id 反查视频服务补媒体 URL（按存储区域选择对应 CDN 域名）、标签、标题（换行符替换为空格，防止破坏 BCP 行结构）；
• IM 私信：按消息类型映射为监管特定枚举，内容分别取文本、语音 URL、图片 URL；
• 支付：最复杂——解析支付单号与金额：微信先尝试解析 XML 回调报文、失败再兼容 JSON（且兼容服务端改动后 array/string 两种结构）、支付宝解析 JSON 字段、苹果与内部充值各有约定；同时映射支付账号类型码。

4.3 落盘前的最后一道校验

bcpDataMapRecord 在拼接 BCP 行之前做两件事：

1. 所有字段值中的换行符替换为空格——BCP 一行一条记录，任何内嵌换行都会破坏行结构、导致后续「按行对齐补列」全部错位；
2. 整行为空白（仅 TAB）则直接丢弃——空行同样会导致补列时行对不上。

5. BCP 文件写入与 Redis 状态管理

5.1 文件命名与滚动

• BCP 文件名类似如（其中一部分）：xxx-xxx-<5位自增序号>-<数据集码>-0.bcp.main（.main 后缀 = 未合并状态标记，序号来自 Redis 自增 hash）；
• 单文件 5000 行上限：每次追加前读行数，超限则生成新文件（新文件名 lpush 进 Redis 文件名 list，表头即当前文件）；
• 追加写用 FileWriter(append) 每条记录一行。

5.2 Redis 中的文件状态键

为什么文件名/计数放 Redis 而不是扫目录：Redis 单线程保证多节点（多 Pod、同 Pod 多消费线程）读到的「当前文件名」一致，这是并发正确性的基石；目录扫描则无法原子。

6. 打包触发：双保险机制

flowchart TB
    subgraph T1["条件触发 - 写入路径内联判断"]
        C1["当前 BCP 满 5000 行"] --> C2{"目录BCP数 ≥ 6 ？"}
        C2 -- 是 --> Z[zipWaFile打包流水线]
        C2 -- 否 --> C3["滚动新文件，继续追加"]
    end
    subgraph T2["定时触发 - Redis 过期事件"]
        D1["每次zipWaFile开始时
setex zip标记300秒"] --> D2["5分钟内没有再打包
key自然过期"]
        D2 --> D3["所有wa Pod的
keyspace notification
监听器收到事件"]
        D3 --> D4{"setnx抢占成功"}
        D4 -- 是 --> Z
        D4 -- 否 --> D5["放弃（已有节点在处理）"]
    end
    Z --> F1["重新setex 300秒
开启下一轮5分钟计时"]

• 条件触发：监管上限是每包 100 个 BCP，工程上主动收紧到 6——刻意让包变小、变频繁，作为「过期事件监听失效」时的兜底，同时按业务体量（视频播放量最大）控制包大小；
• 定时触发：zipWaFile 每次执行的第一步就是 setex 一个 300 秒 TTL 的 Redis key。只要持续在打包，key 不断续期不会过期；一旦 5 分钟没有任何打包动作，key 过期 → 全部 15 个数据集的 Pod 都会收到过期事件（每个 Pod 都监听同一 Redis 的 keyspace notification），保证「至少每 5 分钟产出一包」的时效性约束。

为什么让所有 Pod 都能监听、都能打包：任何一个数据集的 Pod 死掉，其他 Pod 的监听器同样会收到该数据集的过期事件并代为打包——打包能力是全 Pod 对称的，不依赖特定 Pod 存活（监听器里对未知 redisKey 直接忽略，兼容新增数据集时旧版本 Pod 尚未升级的场景）。

多节点抢占：收到过期事件后先 setnx（原子，不能用 get-then-set）抢占 <数据集>_wa_zip 标记，抢到的节点再走文件锁 + 打包；没抢到的直接返回。这里允许一个可容忍的边界：抢占瞬间可能「恰好 key 过期 + 某线程刚 setex 完成正常打包」造成重复 zip，但 zipWaFile 内部会检查目录里是否还有可压缩的 .bcp，已打包过就无文件可打，最坏浪费一个空操作，不影响数据。

7. 并发控制设计（本子系统最核心的部分）

并发来源有四个：同一 Pod 内单流（Kafka 顺序消费，无并发）、多 Pod 并发写同一数据集目录、写入线程与打包流程并发、多个监听器同时收到过期事件。

7.1 数据集级伪分布式文件锁

sequenceDiagram
    participant W as 写入线程（任一 Pod）
    participant R as Redis
    participant F as NAS 上的数据集目录
    W->>R: 自旋等待 file_lock 为空（每 5s 重试）
    W->>R: setex file_lock 300 秒
    Note over W,F: —— 锁内临界区 ——
    W->>R: 读文件名 List 表头 = 当前 BCP
    W->>F: 追加一行数据
    W->>R: 释放锁 del file_lock

• 锁粒度是数据集级（不是全局），不同数据集并行不互相阻塞；
• setex 300 带超时——持有锁的 Pod 崩溃时锁自动释放，不会死锁；
• 锁的直接目的：防止「行数超限 → 生成新文件名写 Redis 前后」的窗口期，两个线程都判断超限、各自生成新文件，产出只有几行数据的碎文件。

7.2 刻意「放弃」的并发场景（工程权衡的典范）

三处不处理、靠后置逻辑兜底的并发：

设计哲学：不追求分布式强一致，而是让每个竞态的后果都落在「可容忍」侧（多一包/超几行/空操作），把真正的互斥只用在数据会错乱的地方（文件锁 + 后缀状态机）。

7.3 文件后缀状态机：重复合并的免疫机制

这是整个并发设计里最巧妙的一环。问题：打包流程（清洗→补列→合并→压缩）是非原子的多步操作，若被打断或并发重入，如何保证不重复处理、不漏处理？

答案：用文件名后缀标记处理阶段，每完成一步就原子改名（mv）推进状态：

flowchart LR
    A["xxx.bcp.main
追加写入中"] -->|"wa_bcp_add.sh 完成合并后
mv 去掉 .main"| B["xxx.bcp
终态，等待打包"]
    A -->|"Java 生成补充列
sidecar 文件"| C["xxx.bcp.main.es.user
补充列数据"]
    C -->|"paste 合并后删除"| B
    B -->|"zip -m 移入压缩包"| D["xxx.zip
最终产物"]

• format/add 脚本只枚举 *.bcp.main——一旦合并完成改名成 .bcp，即使流程被并发重入，也匹配不到已处理文件，天然幂等；
• zip 脚本只收 *.bcp——中间态文件绝无可能混进包；
• 补充列 sidecar 文件（.es.user 等）合并后删除，即使残留也不影响主流程（add 脚本对缺失的补充文件会退化为补空列，见 §8.3）。

7.4 行对齐问题：为什么清洗和补列必须严格按序

补列是按行号横向对齐的（paste），任何一行错位都会导致 IP 归属到错误的用户。因此：

• Java 侧：字段值内嵌换行 → 替换为空格；整行空白 → 整行丢弃（§4.3）；
• shell 侧：先去 NUL 字符再去空行（顺序不可换），并校验首列必须是纯数字 userId，乱数据行整行剔除；
• 这样保证「主文件第 N 行 = 补充文件第 N 行 = 同一个用户」。

8. Java ↔ Shell 混合流水线详解

Java 通过 ProcessBuilder 调用镜像内脚本（CommandUtil 封装：读 stdout/stderr 入日志、waitFor 取退出码；取输出场景用 redirectErrorStream + 全量捕获）。脚本退出码非 0 → Java 抛异常且不重置任何 Redis 状态，下一轮触发可完整重做（配合后缀状态机，已完成的步骤不会重复）。

8.1 wa_bcp_format.sh：三道清洗

对目录下每个 .bcp.main，每道清洗都走「输出到 .bat 临时文件 → 覆盖回原文件」的模式（避免边读边写同一文件）：

1. 去 NUL 字符（sed 's/\x00/ /g'）：^@ 空字节替换为空格。刻意不用 cat -v | tr 方案——cat -v 按字节处理会把中文打乱；
2. 去空行（grep -a -v "^$"）：-a 强制按文本处理，防止含特殊字节时 grep 误判为二进制文件不输出；
3. 首列校验（awk -F'\t' 只保留首列为纯数字或空串的行）：所有 BCP 首列语义固定为 user_id，剔除脏行。

循环内每处理一个文件前都用 find 再确认文件仍存在——应对并发场景下文件可能已被其他线程处理/改名。

8.2 Java 补列：把 ES 画像查询结果变成 sidecar 文件

为什么 IP/经纬度不直接在写入时富化，而要打包前批量补？——写入路径逐条查 ES 开销大且阻塞流，打包时批量查（一次 scroll 查回本包全部 userId 的画像）效率高一个数量级。

流程（genBcpAddData）：

1. 对每个 .bcp.main，调 awk -F'\t' '{print $1}' 提取全部 userId（约定：user_id 必须是 BCP 第一列，CSV 配置时保证）；
2. 按 CSV 推导出该数据集需要的外部源字段（如 es.user.ip、es.user.lng），按对象名分组；
3. 批量查 ES 画像（getSameAttributesByUserIds：<1000 用户 terms 查询，≥1000 scroll 滚动），对 ip 做转长整型、对经纬度做格式化；
4. 按主文件行序生成 <主文件名>.es.user sidecar 文件（行序与主文件严格一致，这是后面 paste 的前提）；
5. 返回「后缀 → 列数」映射，作为下一步 shell 脚本的参数。

CSV 配置有明确约定：同对象名的字段必须在配置中连续排列，保证补充列的列序稳定。

8.3 wa_bcp_add.sh：裁列 + 横向合并

参数契约（由 Java 动态拼装）：脚本 目录 [总列数] [后缀1 列数1] [后缀2 列数2] ...——后缀/列数成对出现、数量可变，意味着多个补充数据源可以任意组合。

对每个 .bcp.main：

1. 裁列（cut -f-N）：按补充列总数从主文件尾部裁掉对应列数，让「主文件保留列 + 补充列 = CSV 定义的总列数」。主文件写入时是按 CSV 全列拼接的（补列字段当时留空占位），裁掉空占位列再贴真实值；
2. 横向合并（paste -d '\t' 主文件 补充文件）：按行号把 sidecar 列贴到主文件右侧；
3. 缺失兜底（awk 补 TAB）：若某个补充文件不存在（异常/已被并发处理），按其列数给每行补空白列——宁可空列也绝不允许列数不对，否则整个 BCP 结构错位；
4. 状态推进：mv xxx.bcp.main xxx.bcp（去掉 .main），并入终态；
5. 循环外删除所有 sidecar 文件。

8.4 索引 XML 生成（Java 侧）

genIndexXml 用 XStream 把对象树序列化为 GAB_ZIP_INDEX.xml，结构为标准报文，包含但不限于：

• 数据交互通用信息
• BCP 文件描述信息
• BCP 数据文件信息：遍历 Redis 文件名清单，逐文件写「文件名 + 实际记录行数」——行数是改名后重新数 .bcp 文件得出的，因此索引行数与包内真实数据严格一致；
• BCP 数据结构：该数据集 CSV 的完整列定义（字段名/英文名/中文含义）——相当于把 Schema 随包自描述；

8.5 wa_zip.sh：打包即清理

单行脚本：zip -q -r -m -P <口令> <zip名> *.bcp GAB_ZIP_INDEX.xml

• 只收 .bcp 终态文件与索引 XML（后缀状态机保证）；
• -m 打包成功后移除源文件——目录里只剩 .zip，即「打包即清理」；
• -P 解压口令（监管侧约定）；

zip 成功后，Java 才重置 Redis 状态：清空文件名 list、删除文件计数——下一轮从全新状态开始。zip 失败则一切状态保留，下轮触发重做（已合并的 .bcp 直接打包，不会重复合并）。

9. 上传层：宿主机 cron + VPN 自愈

wa_ftp_upload.sh 由宿主机 crontab 每分钟执行，流程：

每分钟 cron
  ├─ ① 零点日志轮转：只在 00:00 命中一次，按日期标记截断日志只保留近 3 天
  ├─ ② 清理僵死 ftp 进程：只 kill 处于 S（睡眠）状态的遗留 ftp 子进程，
  │     避免拖住下一轮；绝不 kill 运行态（防止误杀正常上传）
  ├─ ③ VPN 探活：ping 监管侧 FTP 内网地址（1 包/5 秒超时）
  │     ├─ 不通 → docker restart strongSwan VPN 容器（自愈）→ 本轮整体跳过，ZIP 原样保留
  │     └─ 通   ↓
  └─ ④ 遍历 14 个数据集目录：
        目录下有 *.zip 才处理
        → ftp -n -i -v（禁止自动登录 / 关闭 mput 交互 / 详细日志）
        → 显式 user 认证 → bin 二进制模式（防 ZIP 被转码破坏）
        → 远端逐级 mkdir 日期/数据集码/时分 三级目录
          （FTP 无法一次建多级目录；目录已存在 mkdir 报错但流程继续——脚本可安全重入）
        → mput *.zip 批量上传 → bye
        → 上传后删除本地 *.zip（幂等：删了就不会重复传）

传输通道：宿主机上以容器方式常驻 strongSwan（IPSec/IKEv1 主动模式 + PSK 预共享密钥隧道），配置挂载自宿主机 /etc/ipsec.d/，与监管侧内网打通后走 FTP。

自愈设计：VPN 故障时脚本不报错退出，而是重启 VPN 容器后保留全部文件返回——当轮自愈动作 + 下一轮自动恢复上传，无需人工介入。远端按「日期/数据集/时分」分钟级目录隔离批次，天然防覆盖。

10. 存储与部署架构

flowchart TB
    subgraph K8S["Kubernetes - namespace xxx"]
        subgraph WAP["15个 wa* 流应用Pod"]
            P1["wauserregister"]
            P2["wavideoplay"]
            P3["……共 15 个"]
        end
        PVC["PVC: wa-log-pvc
ReadWriteMany / 10Gi
alicloud-nas-sp"]
        P1 & P2 & P3 -- "容器内挂载/data" --> PVC
    end
    NAS[("同一块阿里云 NAS
share 根下的 wa/ 目录")]
    PVC --> NAS
    HOST["宿主机wa_mount.sh:
NFSv3挂载到/xxx
nolock+hard+
1MB读写块"]
    HOST -- "/xxx/wa/<数据集码>" --> NAS
    CRON["crontab每分钟
wa_ftp_upload.sh"]
    CRON --> HOST
    VPN["strongSwan VPN容器
IPSec隧道"]
    CRON -. 探活/重启 .-> VPN

• 同一块 NAS、两种挂载视图：K8s 侧走 PVC（ReadWriteMany 是多 Pod 共写的前提），宿主机侧走 NFSv3 命令式挂载——挂载点不同、底层同一目录，构成「Pod 写 ZIP、宿主机 cron 读 ZIP 上传」的交接面。Pod 刚打好的包对 cron 立即可见，无同步环节；
• Helm 模板用 hasPrefix "wa" 条件判断：仅 wa 前缀应用额外挂载该 PVC，其他流应用不受影响；
• NFS 挂载参数刻意选 nolock（不在 NFS 层做锁，锁职责全部上收到 Redis 应用层）+ hard（NAS 抖动时进程挂起重试而非报错丢数据）；
• 镜像（Alpine JDK8）显式安装 bash（脚本 shebang 需要）与 zip，并做 glibc/UTF-8 locale 处理保证中文不乱码；三个流水线脚本随镜像 COPY 到固定路径供 Java 调用。

11. 关键设计决策汇总

12. 可靠性分析

已具备的保障：

• 不丢数据：追加写 + zip 失败不删源文件（-m 仅成功后移除）+ 状态不重置可重跑；VPN 不通时整轮跳过、文件保留；
• 不重复：后缀状态机防重复合并、setnx 防多节点重复打包、「上传后删除」防重复上传、远端分钟级目录防覆盖；
• 自愈：VPN 容器故障自动重启、流应用假死由 K8s 探针重启（拓扑 RUNNING 状态标记文件）；
• 时效：双触发保证至少 5 分钟一包 + cron 每分钟扫描上传。