一、平台1.0架构

1. 数据源：

目前数据源主要有两类，一类是日志埋点，包括服务端埋点、客户端埋点、H5 埋点，通过 fluentd 服务采集落地到集群磁盘上，另一类是 mysql binlog 数据，通过canal + kafka + fluentd 落地到磁盘上，统一将文件同步到数据集群中(HDFS)

2. 数据分发

主要是数据流转，这里流转的目标主要是两大部分，一部分是流转到离线数据仓库中，一部分是流转到 kafka 中，后续供实时流计算。

2.1 离线仓库

埋点日志

服务端代码通过 fluent api 将原始埋点日志上报至 fluentd 中，fluentd 再将采集数据直接落到服务器磁盘上。

特别说明：

我们封装了两层 fluentd，第一层 fluentd 做 forward 集群，第二层 fluentd 为多台机器多台节点各自独立。服务端是先将埋点日志上报至 forward 集群上，forward 集群再根据配置的不同 match 规则按业务域区分，分别将数据转发至第二层 fluentd 即不同机器节点上的 fluentd。配置类似如

<match cv.product.access.api>
  @type forward
  <server>
    host 10.111.1.x
    port 24224
  </server>
  <buffer>
    flush_interval 3
  </buffer>
</match>

<match cv.*.report.mockrobot>
  @type forward
  <server>
    host 10.111.1.y
    port 24224
  </server>
  <buffer>
    flush_interval 3
  </buffer>
</match>

...

这么做有两个好处：

• 第一层 forward 可做数据缓冲，对于服务端即 fluentd 客户端能做到数据接收入口的统一；
• 第二层按多种不同业务域部署多个 fluentd 到不同的机器节点上，业务域数据物理独立且数据落地磁盘压力被大大降低；

MySQL数据

mysql 的 binlog 数据是通过 canal 服务采集 binlog 会传入到 kafka 中，再通过 fluentd 读取 kafka 中的 binlog 数据落到服务器，binlog -> canal -> kafka -> fluentd -> 磁盘。

# canal.properties 中数据下发 kafka 部分配置
canal.mq.servers=kafka-cluster-kafka-bootstrap.kafka-cluster:9092
canal.mq.acks=1
canal.mq.retries=0
canal.mq.partitionsNum=3
canal.mq.partitionHash=.*\\..*:$pk$

# fluentd-consumer.properties 中 kafka 数据拉取部分配置
fluentd.connect=10.111.1.x:26224

fluentd.tag.prefix=kafka.

fluentd.consumer.topics=binlog-.*

fluentd.consumer.threads=2

fluentd.consumer.backup.dir=/tmp/fluentd-consumer-backup/

fluentd.consumer.from.beginning=true

zookeeper.connect=10.111.1.x:2181,10.111.1.y:2181,10.111.1.z:2181
group.id=binlog-master-fluentd-consumer-group
zookeeper.session.timeout.ms=400
zookeeper.sync.time.ms=200
auto.commit.interval.ms=1000
auto.offset.reset=smallest

这里还需要搭配 Kafka 侧的 Connect 与 Connector 组件服务（Kafka Connect 是一种用于在 Apache Kafka 和外部系统之间传输数据的工具）。

所有经 fluentd 打包并落地磁盘的数据，后续统一通过脚本同步到数仓中（如通过 ossutil 将数据包上传指定 OSS bucket，然后再同步 COS，HDFS 绑定了 COS bucket）。

spark 直接抽取表数据的部分（如 read.jdbc）不属于数据分发，可忽略。

2.2 实时流处理

埋点日志

埋点日志通过 fluentd 落到磁盘后，借助 filebeat 服务 读取 fluentd 的 link 文件 将数据传入到 kafka。

服务端 (Fluent API)
  → 第一层 forward (fluentd-forward/fluent.conf, @type forward)
  → 第二层 aggregator (cvhosts/10.111.1.x/fluentdv1.3/etc/conf.d/td.conf)
       @type file + symlink_path → 写滚动文件 + 生成 .link 符号链接
  → filebeat (tail .link 文件)
  → Kafka topic: report-*

forward 后的第二层 fluentd 类似配置类似如下，持续将数据写入至链接文件中

<match cv.{product,None}.report.video>
  @type file
  @id report_video
  path report_video.log.*.log      # 实际滚动的 chunk 文件
  compress gzip
  symlink_path report_video.link   # ★ 关键：始终指向最新 chunk 文件的符号链接
  <buffer>
    chunk_limit_size 512m
    flush_at_shutdown true
  </buffer>
</match>

symlink_path 的工作方式：file 插件每次 flush 出一个新的 chunk 文件（report_video.log.<时间戳>.<hash>.log.gz）后，会原子地更新这个符号链接指向最新文件。这样 filebeat 只需要盯住一个固定路径 report_video.link，就能连续读到所有数据。这就是 fluentd → filebeat 的标准对接方式（官方文档里 symlink_path 的设计用途就是 “hook up with tail-like programs”）。

filebeat 侧类似配置如下

filebeat.inputs:
- type: log
  paths:
  - /logs/report_video.link    # 盯住 fluentd 生成的 symlink
  symlinks: true               # 允许跟随符号链接
  tail_files: true             # 启动时从文件末尾开始，不重读历史
  processors:
  - dissect:                   # 按制表符切开 fluentd 默认的 out_file 行格式
      tokenizer: "%{time}\t%{tag}\t{%{record}"
      target_prefix: "log"
output.kafka:
  hosts: ["kafka-cluster-kafka-brokers.kafka-cluster:9092"]
  topic: 'report-video'
  version: 1.1.0
  required_acks: 1
  client_id: beat_report_video
  codec.format:                # 把切开的字段重新拼成合法 JSON 发出去
    string: '{"timestamp":"%{[log.time]}",%{[log.record]}'

其实早期并没有额外引入 filebeat，而是 fluentd 直接到 kafka，fluentd 中类似配置如下

# fluentd 中部分配置
@type kafka_buffered
brokers kafka-host:9092
default_topic report-video
flush_interval 3s
retry_limit 1
buffer_type file
buffer_path /data5/logs/cv/report_video.log.kafka

kafka_buffered：老版本 fluent-plugin-kafka（0.x 时代）的输出插件，自带持久化 buffer。现在已废弃，新版插件里统一叫 kafka2（kafka_buffered 的参数它全兼容）。

• 优点：链路短，少跳一组件；fluentd kafka 插件自带 buffer、重试、ack 控制；
• 缺点：fluentd 直接面对 Kafka 可用性时要有足够大的 buffer（磁盘 buffer）防止 Kafka 抖动丢数据。

而 filebeat 有完善的 registry/断点续传，进程挂了重启不丢已落盘的日志，同时 fluentd 和 Kafka 解耦。尽管多一个组件运维且多一次磁盘 I/O，但可忽略不计。

MySQL数据

binlog 则通过 canal 直接出入到 kafka 中  后续统一通过 kafka 进行实时流式计算。

3. 数据处理

主要两部分： 离线数仓+实时流

离线数仓：构建数据仓库，主要使用 Spark DataFrame、Hive SQL 进行 ETL 清洗，数据保存到 EMR 集群中

实时流：主要使用 KafkaStream 进行 Kafka 流的处理。

4. 数据存储

处理后的数据主要保存在 HDFS 上个别应用数据会回写给业务，保存到 Redis 或者数据库中， 部分还会保存在 ES 中，一些指标数据会写入Influxdb 中，通过 Grafana 进行展示。

后端系统部署在阿里云内，但数据平台因为早期数据平台因成本等多因素考虑部署在腾讯云侧，所以尤其离线数据在链路末端先是存于 OSS 上，再通过定制化脚本和服务同步至 COS 上，最终集成挂载于 HDFS 中。

5. 组件

• HDFS + Yarn + Spark + Hive
• Airflow：DAG ELT数据清洗任务调度系统
• Hue
• Zeppelin：日常主要的看板与数据查询平台，主要针对离线数据，提供给运营、产品、数据分析师等多方人员。
• Ranger：没有使用自建 ldap，hdfs 与 zeppelin 使用了 ranger 进行权限控制
• Knox：安全反向代理与 API 网关系统
• Grafana：日常实时数据大盘与组件服务监控

二、平台迁移与演进

后来经整体考量将腾讯云 EMR 迁移阿里云 EMR，深度使用了阿里云 EMR 产品。

对于数据存储来说，阿里云 EMR HDFS 中直接挂载 OSS，数据无需再同步腾讯云 COS。

1. 数据分发

因为某些契机我们舍弃了 Kakfa，改由阿里云 SLS 代替，也完全废弃了 fluentd。

后端直接通过 SLS api 上报埋点日志至 SLS 中，MySQL binlog 也由 canal 采集改为 logtail，将 canal→ kafka → fluentd → oss → cos 收集链路替换为 logtail→ sls → flink → oss。

离线数仓 OSS 数据由 Flink 消费 SLS 落地。

2. 数据处理

离线数据清洗，原先大量的 Spark Scala 实现改为 Spark SQL 或 Hive SQL 的方式。

实时流计算改为 Flink + SLS Stream：

• 绝大部分实时数据清洗由 Flink 直接完成
• 监管数据实时上报网警因为涉及多组件交互（如 FTP）、逻辑复杂（需要 Shell 脚本执行）且监控告警等，故无法 Flink 完成，故由原先 Kafka Stream 改为 SLS Stream，采用 SLS 原生 SDK loghub-client 完成，这样迁移再开发工作量大大减少。

实时数据查询引入了 Hologress 的支持，日常开发 HoloWeb 或其他 BI 工具直接查询，也增加了其他端口的支持，如：

• 公司企业后台提交固定业务的查询请求（固定SQL），对应后端接口内 Holo JDBC 查询；
• Zeppelin Holo 解释器查询，提供给运营产品岗对榜单业务数据的实时获取。

3. 组件

• 废弃了 Ranger，改为 LDAP 集成；
• 原先 Zeppelin 做为对运营、产品等非开发岗的主要数据查询展示平台外，还引入了 Superset 搭配使用；
• Airflow 调度先是迁移到了阿里云 数据开发控制台 系统中，因 2025 年开始下线，后迁移到阿里云 Workflow 上（EMR Workflow是一个全托管的工作流和任务调度服务，100%兼容开源Apache DolphinScheduler 海豚调度）；

4. 数仓规范

离线数仓分层结构进行重构，将原先 idl、cdl、adl 等分层改为了 ods、dim、dwd、dws、ads 的结构。