数据实时流处理平台（Kafka Streams）项目介绍

实时数仓/实时数据流处理平台（Kafka + Kafka Streams + K8s）：从 0 到 1 搭建 App 实时数仓，以「单镜像 + Spring Profile」模式在 K8s 上裂变 40+ 个独立流处理应用，消费埋点与 Canal CDC 两类 Kafka 数据源，产出 10 分钟粒度实时经营看板（InfluxDB）、秒级用户画像（ES 宽表）、实时风控反引流（Redis 频控 + 内容/图片识别 + 自动封禁）、热门内容 TopN 榜单、广告归因回流与监管合规上报；解决窗口最终值结算、at-least-once 幂等、大基数 UV 去重、黑产字符混淆对抗等核心问题，稳定服务百万级用户 5 年。

1. 项目概述

项目名称：App 数仓实时计算平台

项目背景：App 是一款面向中老年群体的短视频图文创作与社交社区的产品（视频/直播/IM 私信/充值打赏），业务上对数据的实时性有四类核心诉求：

1. 实时经营看板——活跃用户/设备、注册、直播、推送、支付、资产等核心指标以 10 分钟粒度产出，供运营与管理层实时决策；
2. 实时用户画像——把行为日志与数据库变更秒级汇聚成用户宽表画像，支撑推荐、风控与运营查询；
3. 实时风控反作弊——对引流导流、批量注册、刷量等黑产行为做秒级发现与自动处置（封禁/验证码）；
4. 合规与外部数据回流——按监管规范上报明细行为数据（BCP 文件 + FTP），并向广告投放平台回传激活/留存/付费事件用于投放优化。

项目定位：一套以 Kafka 为数据总线、以 Kafka Streams 为计算引擎的实时数仓/实时处理平台，覆盖了实时 ETL、窗口聚合、CDC 增量同步、实时画像、流式风控、榜单计算、合规上报等典型实时场景。

规模量级：

• 单一代码库裂变为 40+ 个独立部署的流处理应用，覆盖 20+ 个 Kafka 数据源 Topic（埋点类 + CDC 类）；
• 服务百万级注册用户的实时指标计算与画像更新；
• 代码约 150 个 Java/Scala 类，持续迭代 5 年以上（2018–2023），日均处理全量埋点与数据库变更流。

2. 整体架构

2.1 架构总览图

flowchart TB
    subgraph SRC["数据源层"]
        A1["App 客户端服务端埋点
report-video / report-im / report-push 等"]
        A2["网关 API 访问日志
access / access-api"]
        A3["业务 MySQL 集群
用户库 / 资金库 / 审核库 / 话题库"]
    end

    subgraph MQ["采集与消息层"]
        C1["Fluentd -> filebeat"]
        C2["Canal
MySQL Binlog 转 JSON FlatMessage"]
        B[("Kafka 集群
埋点 Topic 与 binlog Topic
及 Streams 状态 Changelog")]
    end

    subgraph RT["实时计算层 - Kafka Streams 共 40+ 流应用"]
        D1["实时统计看板
10 分钟窗口聚合"]
        D2["用户画像
多源 upsert"]
        D3["实时风控
频次 / UV / 内容 / 图片"]
        D4["实时榜单 TopN
滑动窗口"]
        D5["广告归因回流
多平台回调"]
        D6["合规上报 WA
BCP 文件"]
        D7["DataFinder 回流
攒批上报"]
    end

    subgraph STORE["存储与服务层"]
        E1[("InfluxDB
十分钟看板")]
        E2[("Elasticsearch
user_profile 画像")]
        E3[("PostgreSQL
风控与统计数据")]
        E4[("Redis
去重 / 锁 / 榜单 / 计数")]
        E5[("MySQL
业务库与离线数仓")]
        E6["推荐服务 / 审核服务
反垃圾服务 / 封禁服务"]
        E7["广告平台回调
头条 / 快手 / 腾讯"]
        E8["监管 FTP 上报
BCP 与 ZIP"]
        E9["火山 DataFinder"]
        E10["钉钉告警机器人"]
    end

    A1 --> C1
    A2 --> C1
    C1 --> B
    A3 --> C2 --> B
    B --> D1
    B --> D2
    B --> D3
    B --> D4
    B --> D5
    B --> D6
    B --> D7
    D1 --> E1
    D2 --> E2
    D3 --> E3
    D3 --> E4
    D3 --> E6
    D3 --> E10
    D4 --> E4
    D5 --> E7
    D6 --> E8
    D7 --> E9
    D2 --> E5
    D1 -. 批量写失败或流异常 .-> E10

2.2 部署架构图

flowchart LR
    subgraph CICD["CI 与 CD 流水线"]
        S1["Maven 构建
单一 Fat Jar"] --> S2["Docker 镜像
streams-appName:tag"] --> S3["Helm 与 K8s"]
    end
    subgraph K8S["Kubernetes 集群 - namespace colorv"]
        subgraph POOL["大数据专用节点池 - nodeSelector big.data=stream"]
            P1["Pod: mau-stream
profile=mau"]
            P2["Pod: suspectim-stream
profile=suspectim"]
            P3["Pod: hotaudio-stream
profile=hotaudio"]
            PN["…… 共 40+ 个独立 Deployment"]
        end
        CM["ConfigMap
生产配置热刷新"] -.-> P1
        CM -.-> P2
        CM -.-> P3
        NAS[("NAS PVC
合规文件共享盘")] -.-> PN
    end
    S3 --> P1
    S3 --> P2
    S3 --> P3
    S3 --> PN

部署模式核心思想：一个代码库、一个 Fat Jar、一个通用镜像，借 Spring Profile 在 K8s 上裂变为几十个独立的流处理应用。

• 每条流 = 一个 @Configuration + @EnableKafkaStreams + @Profile("xxx") 的拓扑声明类；
• 容器启动命令固定为 java -jar -Dspring.profiles.active=prod,<appName> app.jar，一个 Deployment 只激活一个拓扑；
• 每个应用以 STREAM_NAME=<appName>-stream 作为 Kafka Streams 的 application.id → 独立消费组、独立状态存储（RocksDB State Store + Changelog）、独立扩缩容与滚动升级，单条流的故障/重启完全不影响其他流；
• Helm Chart 用一个模板遍历 values.yaml 中的应用列表批量生成 Deployment，构建脚本支持按应用名单独构建、发版、灰度滚动；
• 资源规格：单 Pod 限制 1C/2G，调度到专用大数据节点，关闭 Service Mesh sidecar 注入以降低流处理时延。

3. 技术栈

4. 核心原理与逻辑（按数据链路展开）

4.1 数据接入层：两类数据源

Canal 消息解析模型（CDC 链路的统一入口）：每条消息携带 database/table（库表路由）、type（INSERT/UPDATE/DELETE）、data（变更后行数据数组）、old（变更前列值）、es（MySQL 执行时间）与 ts（Canal 处理时间）。计算层统一模式为：JsonSerde(FlatMessage) 反序列化 → 按 库.表 过滤路由（branch 分流）→ flatMapValues(getData()) 摊平多行 → 按事件类型处理。其中 data 与 old 的差分是资产增量对账、订单状态迁移判断（如「支付回调字段由 0 变非 0」）的关键。

4.2 实时统计看板（十分钟看板）

模式：stream → filter/map → groupByKey → windowedBy(10 分钟滚动窗口) → count/reduce/aggregate(Materialized 命名状态库) → foreach 写 InfluxDB。

典型指标与去重手段：

4.3 实时用户画像

flowchart LR
    subgraph 多源数据
        B1[access-api 访问日志] -->|ip| M
        B2[report-video 地理事件] -->|省市/经纬度| M
        B3[binlog 手机账号表] -->|phone| M
        B4[binlog 实名认证表] -->|姓名/证件| M
        B5[binlog 资金库资产表] -->|钻石/金币/金豆| M
        B6[access-api 老用户回访] -->|last_active/app_version/渠道| M
    end
    M[ES user_profile 索引
doc_id = user_id
doc 局部更新 + 冲突重试]
    M --> R1[风控：用户价值/风险判定]
    M --> R2[合规上报：IP/经纬度/行政区划补全]
    M --> R3[运营查询]

• 宽表式画像：不建大宽表 Schema，而是各维度独立事件流按 user_id 持续 upsert 到同一 ES 文档（UpdateRequest.doc 局部合并 + retryOnConflict），单维度更新互不覆盖、天然幂等可重放；
• 「最后活跃/版本/渠道」类弱实时字段用 按天窗口 + reduce 取最新值 的方式降频更新（每用户每天至多一条）；
• 读取侧按规模自适应：<1000 用户走 terms 查询，≥1000 走 scroll 滚动查询，多用户多字段用 msearch 批量。

4.4 实时风控（反引流/反作弊）

风控是本项目逻辑最复杂、对抗性最强的部分。总体架构为「5 路信号采集 → 统一嫌疑服务 → 分级处置」：

flowchart TB
    subgraph SIG["五路信号采集 - 各自独立流应用"]
        S1["接口频次
access-api 高危接口"]
        S2["IM 私信
腾讯云回调"]
        S3["评论频次
report-video"]
        S4["文本内容
评论 / 昵称 / 签名 / 私信"]
        S5["图片内容
头像 / 相册 / 私信图 / 话题封面"]
    end
    subgraph CORE["统一判定 SuspectContentService"]
        J1["字符反混淆归一化
花体数字 / 全角 / 形近字 / 中文数字"]
        J2["正则抽取嫌疑关键词
微信号 / QQ / 手机号 / URL"]
        J3["图片 hash 黑白名单
与 OCR 图中文字"]
        J4["PG 幂等落库
key 加 user 加 md5 唯一，count 累加"]
        J5["阈值分级
全站与单用户两级阈值"]
    end
    subgraph ACT["分级处置"]
        O1["验证码挑战
弱信号 - 评论频次"]
        O2["系统级全站封禁
强信号 - 高危渠道包 / 黑词 / 黑图"]
        O3["钉钉告警与黑词库回灌
人工在回路定性"]
    end
    S1 --> J4
    S2 --> J4
    S3 --> J4
    S4 --> J1
    J1 --> J2
    J2 --> J4
    S5 --> J3
    J3 --> J4
    J4 --> J5
    J5 --> O1
    J5 --> O2
    J5 --> O3

关键设计点：

1. Redis Key + TTL 即风控窗口：分钟级（Set 统计私信触达的不同收件人 UV，阈值 60）、日级（UV 500 / 接口 PV 2000/1000/500 按接口分级）——风控计数不依赖 Kafka Streams 窗口，跨实例共享、天然滚动过期；
2. 新老用户水位线：用自增 user_id 水位（如「3 个月前最小 user_id」）区分新老账号，老用户豁免频控，聚焦新注册黑产小号，同时结合设备/渠道画像（特定老安卓版本 + 应用商店渠道包）直接封禁；
3. 「先观测、后定性、再自动封禁」的人在回路机制：新出现的联系方式只累计不封禁；被运营/告警定性为黑产的词/图进入黑名单后，再传播者自动封禁（图片用七牛 hash 跨用户复用已定性违规图）；普通内容要求单用户历史累计超阈值才封，避免误伤；
4. 文本反混淆对抗：黑产用花体 Unicode 数字、全角字符、中文数字、形近字拼联系方式，系统先做字符归一化再正则抽取（微信号样式 / 5–20 位连续数字 / URL），长文本按 100 字切片送检防截断漏匹配；
5. 行为宽表：另有一路把「每用户每小时对多少不同对象点赞/评论/关注/私信」的去重 UV 写入 PG 小时表，供离线策略分析。

4.5 实时榜单 TopN（热门音乐/搜索热词/场景词）

两段式窗口聚合 + 自研 Top 容器 + 原子发布：

第一层：按内容 key 分组 → 6 小时窗口 / 1 小时滑动 hopping window
        → aggregate 出 (曝光数, 使用数)          [WindowStore]
第二层：以「窗口结束时间」为 key → 聚合进自研 Top<T> 容器
        （TreeSet 定长有序、支持窗口滑出时撤回 remove）  [KeyValueStore]
输出：  仅发布「窗口结束距今 1 小时内」的榜单元数据
        → Redis 临时 list 写满后 rename 原子替换正式 key

• 自研 Top<T> 容器解决「TopN 状态需要可序列化恢复 + 窗口过期撤回」两个问题，配套自定义 Deserializer 在状态恢复时重建容器；
• 榜单排序以真实使用量为主权重、曝光量为次序，防刷量；
• 临时 key + rename 保证读侧永远看不到半成品榜单。

4.6 CDC 增量对账（全站资产总额统计）——最典型的窗口工程难点

问题：Kafka Streams 的 windowedBy(...) 后的输出不是窗口关闭时触发一次，而是以秒级节奏持续增量触发；且把「新增」和「减少」拆成两条支流后，并发回读时序库上一窗口值存在竞态。

解法（当前窗口只累加、下一窗口做结算，Redis 单线程做仲裁）：

sequenceDiagram
    participant K as Kafka Streams 窗口聚合
    participant R as Redis（单线程仲裁）
    participant I as InfluxDB
    Note over K: 窗口 N（10 分钟）
    K->>R: add 流：data 新值窗口求和 → key add_N
    K->>R: sub 流：old 旧值窗口求和 → key sub_N
    Note over K: 进入窗口 N+1（start == N.end 对齐）
    K->>R: setnx 占位（保证只结算一次）
    K->>R: 读 final_N-1 + add_N − sub_N
    K->>I: 写入全站资产总额（10 分钟粒度）
    K->>R: 更新 final_N，清理 N 周期中间 key

配合「实时先行、离线 T+1 校正差值」的口径对账思路，解决了实时值与离线基准的衔接。同一模式（窗口对齐 + Redis 缓存 + 下一窗口结算）也用于「视频发布→审核入库平均时长」等均值类指标。

4.7 广告投放归因与事件回流

归因链路：广告点击日志（含设备标识 only_key / idfa / udid + 计划/创意/广告主/投放端维度）落业务库 → 实时流以设备标识反查点击记录富化投放维度（内置投放端/广告主/代理商字典翻译）→ 按维度窗口计数产出「激活归因」「新客当天创作转化」实时指标。

事件回流链路（激活/次留/付费回传给投放平台，用于平台侧 oCPX 优化）：

• 次留回流：识别「昨日首次访问、今日仍活跃」的用户 → 查 Redis 中缓存的该设备回调 URL → 按回调 URL 域名路由到不同平台（头条/快手/腾讯/友盟），以各平台约定的 event_type 组装参数回传；
• 付费回流：监听订单表 binlog 的支付成功状态迁移（old/data 差分判定）→ 回传付费事件；
• 幂等保障：回调成功即删除 Redis 回调缓存 key，确保每台设备每类事件只回传一次；失败钉钉告警。

4.8 监管合规数据上报（WA 链路）

按监管规范上报 15 类明细行为数据（注册/登录/好友/播放/评论/点赞/关注/分享/私信/支付/充值/下单/打赏/搜索），每条记录需补全行政区划（GB/T 2260 六位码，本地字典 + 画像省市区转换）、IP（转长整型）、经纬度等溯源字段。

链路设计（典型「流式落盘 + 定时打包上传」架构）：

Canal binlog / 埋点 topic
  → 流内过滤富化（CSV 配置驱动的「监管字段 ← 业务库.表.字段」映射，非硬编码）
  → 追加写 BCP 数据文件（NAS 共享盘，多 Pod 文件锁防并发；单文件 5000 行滚动切分）
  → 凑够 N 个文件 或 Redis 触发 key 过期（keyspace notification 监听，保证至少每 5 分钟一包）
      → setnx 抢锁（多节点只有一个执行）
  → shell 清洗 + 从 ES 画像批量补列 + 横向合并
  → 生成索引 XML（GAB_ZIP_INDEX，XStream 序列化）→ 打 ZIP
  → 宿主机 crontab 每分钟经 VPN + FTP 上传监管侧（上传前自动检测 VPN 健康、失败自愈重启）

4.9 服务端埋点回流（DataFinder）

把服务端日志与订单 binlog 二次加工成标准埋点事件（user/header/events 三段式协议），回流到增长分析平台，补齐客户端埋点覆盖不到的服务端行为（VIP/钻石商品下单与支付成功等商业化事件）。

• 10 秒短窗口攒批：按 userId 分组、窗口内字符串拼接聚合成批后批量上报，显著降低外呼请求数（批量接口有条数上限）；
• 10 线程异步 + @Retryable 指数退避重试，失败钉钉告警；
• 客户端事件按「事件 ID 白名单（Redis Set 动态配置）」过滤后回流。

4.10 实时推荐协同

推荐算法本体在外部 Python 推荐服务，本平台承担实时喂料与门槛判定：

• 话题/图文内容 binlog → 过滤（审核通过/未删/白名单作者/正文长度）→ HTTP 调推荐服务加入随机池；运营标记上榜 → 加入精选池；删除/下架 → 移出；每个动作成败打点监控；
• 用户登录 → 清空推荐会话（避免重复曝光）；
• 对外暴露「视频是否可推荐」判定 HTTP 接口：视频存在性 + 非模板默认标题 + 作者白名单 + 封面图像质量判定（调用图像服务检测黑边/画质）四重门槛。

5. 关键技术难点与解决方案

6. 项目成果与价值

• 建成覆盖「埋点 + CDC」双数据源的统一实时数仓，支撑活跃/直播/推送/支付/资产/审核时效/投放等全业务线 10 分钟粒度实时看板，替代 T+1 离线报表成为运营决策主入口；
• 秒级用户画像（ES 宽表）成为推荐、风控、运营的统一用户数据底座；
• 实时风控体系（五路信号 + 分级处置 + 人在回路黑名单）有效压制中老年社区高发的引流导流黑产，支持验证码/系统封禁自动处置；
• 打通广告投放归因—回流闭环（激活/次留/付费回传多平台），直接服务买量优化；
• 完成监管合规上报工程建设（15 类数据集、BCP/ZIP/FTP/VPN 全链路自动打包上传），满足监管审计要求；
• 平台稳定运行 5 年，后期主导按业务优先级分批下线与迁移（埋点链路迁移至日志服务方案），完成技术栈的新老交替收尾。