我们前面采集的日志数据已经保存到 Kafka 中，作为日志数据的 ODS 层，从 Kafka 的ODS 层读取的日志数据分为 3 类, 页面日志、启动日志和曝光日志。这三类数据虽然都是用户行为数据，但是有着完全不一样的数据结构，所以要拆分处理。将拆分后的不同的日志写回 Kafka 不同主题中，作为日志 DWD 层。
页面日志输出到主流,启动日志输出到启动侧输出流,曝光日志输出到曝光侧输出流
3.1 主要任务
3.1.1 识别新老用户
本身客户端业务有新老用户的标识，但是不够准确，需要用实时计算再次确认（不涉及业务操作，只是单纯的做个状态确认）。
3.1.2 利用侧输出流实现数据拆分
根据日志数据内容，将日志分为3类，页面日志，启动日志和曝光日志。页面日志输出到主流，启动日志输出到启动侧输出流，曝光日志输出到曝光日志侧输出流
3.1.3 将不同流的数据推送下游的kafka的不同topic中
3.2 代码实现
3.2.1 接收kafka数据，并进行转换
1）在Kafka的工具类中提供获取kafka消费者的方法（读）
public class KafkaUtil {private static String KAFKA_SERVER = "node001:9092,node002:9092,node003:9092";private static Properties properties = new Properties();static {properties.setProperty("bootstrap.servers", KAFKA_SERVER);}/*** 获取KafkaSink的方法* @param topic 主题*/public static FlinkKafkaProducer<String> getKafkaSink(String topic) {return new FlinkKafkaProducer<String>(topic, new SimpleStringSchema(), properties);}/*** 获取KafkaSource的方法* @param topic 主题* @param groupId 消费者组*/public static FlinkKafkaConsumer<String> getKafkaSource(String topic,String groupId){//给配置信息对象添加配置项properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);//获取KafkaSourcereturn new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),properties);}}
2）Flink 调用工具类读取数据的主程序
package com.djin.gmallrealtime.app;import akka.stream.impl.FailedSource;import com.alibaba.fastjson.JSONObject;import com.djin.gmallrealtime.utils.KafkaUtil;import org.apache.flink.runtime.state.filesystem.FsStateBackend;import org.apache.flink.streaming.api.CheckpointingMode;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;/*** Flink调用工具类读取数据的主程序*/public class BaselogApp {public static void main(String[] args) throws Exception {// 1.获取执行环境，设置并行度，开启CK，设置状态后端（HDFS）StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 为Kafka主题的分区数env.setParallelism(1);// 1.1 设置状态后端// env.setStateBackend(new FsStateBackend("hdfs://node001:8020/gmall/dwd_log/ck"));// 1.2 开启CK// env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE);// env.getCheckpointConfig().setCheckpointTimeout(60000L);// 修改用户名System.setProperty("HADOOP_USER_NAME","djin");// 2.读取Kafka ods_base_log 主题数据String topic = "ods_base_topic";String groupId = "ods_dwd_base_log_app";FlinkKafkaConsumer<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);// 3.将每行数据转换为JsonObjectSingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSONObject::parseObject);// 打印测试jsonObjDS.print();// 执行任务env.execute();}}
// 4. 按照Mid分组KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(data ->data.getJSONObject("common").getString("mid"));// 5. 使用状态做新老用户校验SingleOutputStreamOperator<JSONObject> jsonWithNewFlagDS = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {// 声明状态用于表示当前Mid是否已经访问过private ValueState<String> firstVisitDateState;private SimpleDateFormat simpleDateFormat;@Overridepublic void open(Configuration parameters) throws Exception{firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("new_mid", String.class));simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");}@Overridepublic JSONObject map(JSONObject value) throws Exception {// 取出新用户标记String isNew = value.getJSONObject("commen").getString("is_new");// 如果当前前端传输数据表示为新用户，则进行校验if ("1".equals(isNew)){// 取出状态数据并取出当前访问时间String firstDate = firstVisitDateState.value();Long ts = value.getLong("ts");// 判断状态数据是否为Nullif (firstDate != null) {// 修复value.getJSONObject("common").put("is_new", "0");} else {// 更新状态firstVisitDateState.update(simpleDateFormat.format(ts));}}return value;}});// 打印测试jsonWithNewFlagDS.print();
3.2.3 利用侧输出流实现数据拆分
根据日志数据内容，将日志数据分为3类，页面日志，启动日志和曝光日志。页面日志输出到主流，启动日志输出到启动侧输出流，曝光日志输出到曝光侧输出流
// 6.分流，使用ProcessFunction将ODS数据拆分成启动、曝光以及页面数据
SingleOutputStreamOperator<String> pageDS = jsonWithNewFlagDS.process(
new ProcessFunction<JSONObject,String>() {
@Override
public void processElement(JSONObject value,
Context ctx,
Collector<String> out) throws Exception {
// 提取”start“字段
String start = value.getString("start");
// 判断是否为启动数据
if (start != null && start.length() > 0) {
// 将启动日志输出到侧输出流
ctx.output(new OutputTag<String>("start") {
},
value.toString());
} else {
// 为页面数据，将数据输出到主流
out.collect(value.toString());
// 不是启动数据，继续判断是否是曝光数据
JSONArray displays = value.getJSONArray("displays");
if (displays != null && displays.size() > 0) {
// 为曝光数据，遍历写入侧输出流
for (int i = 0; i < displays.size(); i++) {
// 取出单条曝光数据
JSONObject displayJson = displays.getJSONObject(i);
// 添加页面ID
displayJson.put("page_id",
value.getJSONObject("page").getString("page_id"));
// 输出到侧输出流
ctx.output(new OutputTag<String>("display") {
},
displayJson.toString());
}
}
}
}
}
);
// 7.将三个流的数据写入对应的Kafka主题
DataStream<String> startDS = pageDS.getSideOutput(new OutputTag<String>("start") {});
DataStream<String> displayDS = pageDS.getSideOutput(new OutputTag<String>("display") {});
pageDS.addSink(KafkaUtil.getKafkaSink("dwd_page_log"));
startDS.addSink(KafkaUtil.getKafkaSink("dwd_start_log"));
displayDS.addSink(KafkaUtil.getKafkaSink("dwd_display_log"));
// 打印测试
pageDS.print("Page>>>>>>");
startDS.print("Start>>>>>>");
displayDS.print("Display>>>>>>");
3.2.4 将不同流的数据推送到下游kafka不同的topic(分流)
1）程序中调用Kafka工具类获取Sink
pageDS.addSink(KafkaUtil.getKafkaSink("dwd_page_log"));
startDS.addSink(KafkaUtil.getKafkaSink("dwd_start_log"));
displayDS.addSink(KafkaUtil.getKafkaSink("dwd_display_log"));
2）测试
- IDEA中运行BaseLogApp类
- 运行logger.sh，启动Nginx以及日志处理服务
- 运行rt_applog下模拟数据的jar包
- 到kafka不同的主题下查看输出效果
若有收获，就点个赞吧
0 人点赞
