小编点评

**日志数据拆分处理方案** **1. 页面日志** - 从 Kafka 中读取页面日志数据。 - 将页面日志数据拆分成页面视图和具体页面的信息。 - 将页面视图和页面信息写入不同的 Kafka 主题中。 **2. 启动日志** - 从 Kafka 中读取启动日志数据。 - 将启动日志数据处理并将其写入不同的 Kafka 主题中。 **3. 曝光日志** - 从 Kafka 中读取曝光日志数据。 - 将曝光日志数据处理并将其写入不同的 Kafka 主题中。 **步骤：** 1. 将页面日志、启动日志和曝光日志从 Kafka 中读取出来。 2. 将日志数据拆分成页面视图和页面信息。 3. 将页面视图和页面信息写入不同的 Kafka 主题中。 4. 将启动日志和曝光日志写入不同的 Kafka 主题中。 5. 创建一个新的 Kafka 流，从 `dwd_base_log` 主题中读取数据并将其转换为 JSON 对象。 6. 将每行数据转换为 JSON 对象并写入不同的 Kafka 主题中。 7. 创建一个新的 Kafka 流，从 `dwd_page_log、dwd_display_log` 主题中读取数据并将其处理。 8. 将页面日志、启动日志和曝光日志输出到不同的 Kafka 主题中。 **注意：** - 每个主题的名称应应与日志类型命名相关。 - 每个主题的配置应根据实际需求进行调整。 - 可以使用 Flink CDC 创建动态的日志数据管道，自动处理日志数据拆分和写入。

正文

我们前面采集的日志数据已经保存到 Kafka 中，作为日志数据的 ODS 层，从 Kafka 的ODS 层读取的日志数据分为 3 类, 页面日志、启动日志和曝光日志。这三类数据虽然都是用户行为数据，但是有着完全不一样的数据结构，所以要拆分处理。将拆分后的不同的日志写回 Kafka 不同主题中，作为日志 DWD 层。
流页面日志输出到主流,启动日志输出到启动侧输出流,曝光日志输出到曝光侧输出流

识别新老用户

本身客户端业务有新老用户的标识，但是不够准确，需要用实时计算再次确认(不涉及业务操作，只是单纯的做个状态确认)。

启动日志

曝光日志

页面日志

实现逻辑

• 获取执行环境
• 消费 ods_base_log 主题数据创建流
• 将每行数据转换为JSON对象(脏数据写到侧输出流)
• 新老用户校验 状态编程
• 分流 侧输出流 页面：主流 启动：侧输出流 曝光：侧输出流
• 提取侧输出流
• 将三个流进行打印并输出到对应的Kafka主题中
• 启动任务

# 启动三个消费者，分别消费 dwd_start_log、dwd_page_log、dwd_display_log 主题
$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic dwd_page_log

尚硅谷 源代码
B站视频 DWD&DIM

大数据-数据仓库-实时数仓架构分析
大数据-业务数据采集-FlinkCDC
大数据 - DWD&DIM 行为数据
大数据 - DWD&DIM 业务数据
大数据 DWM层 业务实现