1. 摘要

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

页面日志输出到主流，启动日志输出到启动侧输出流，曝光日志输出到曝光侧输出流

2. 识别新老用户

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

利用侧输出流实现数据拆分

根据日志数据内容,将日志数据分为 3 类：页面日志、启动日志和曝光日志。将不同流的数据推送下游的 kafka 的不同 Topic 中

3. 代码实现

在包app下创建flink任务BaseLogTask.java，

通过flink消费kafka 的数据，然后记录消费的checkpoint存到hdfs中，记得要手动创建路径，然后给权限

checkpoint可选择性使用，测试时可以关掉。

package com.zhangbao.gmall.realtime.app;
import com.alibaba.fastjson.JSONObject;
import com.zhangbao.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
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;
/*** @author: zhangbao* @date: 2021/6/18 23:29* @desc:**/
public class BaseLogTask {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//设置并行度，即kafka分区数env.setParallelism(4);//添加checkpoint，每5秒执行一次env.enableCheckpointing(5000, CheckpointingMode.EXACTLY\_ONCE);env.getCheckpointConfig().setCheckpointTimeout(60000);env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));//指定哪个用户读取hdfs文件System.setProperty("HADOOP\_USER\_NAME","zhangbao");//添加数据源String topic = "ods\_base\_log";String group = "base\_log\_app\_group";FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, group);DataStreamSource<String> kafkaDs = env.addSource(kafkaSource);//对格式进行转换SingleOutputStreamOperator<JSONObject> jsonDs = kafkaDs.map(new MapFunction<String, JSONObject>() {@Overridepublic JSONObject map(String s) throws Exception {return JSONObject.parseObject(s);}});jsonDs.print("json >>> --- ");
try {//执行env.execute();} catch (Exception e) {e.printStackTrace();}
}
}

MyKafkaUtil.java工具类

package com.zhangbao.gmall.realtime.utils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;
/*** @author: zhangbao* @date: 2021/6/18 23:41* @desc:**/
public class MyKafkaUtil {private static String kafka_host = "hadoop101:9092,hadoop102:9092,hadoop103:9092";/*** kafka消费者*/public static FlinkKafkaConsumer<String> getKafkaSource(String topic,String group){Properties props = new Properties();props.setProperty(ConsumerConfig.GROUP\_ID\_CONFIG,group);props.setProperty(ConsumerConfig.BOOTSTRAP\_SERVERS\_CONFIG,kafka_host);return new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(),props);}
}

4. 新老访客状态修复

识别新老客户规则

识别新老访客，前端会对新老客状态进行记录，可能不准，这里再次确认，保存mid某天状态情况（将首次访问日期作为状态保存），等后面设备在有日志过来，从状态中获取日期和日志产生日期比较，如果状态不为空，并且状态日期和当前日期不相等，说明是老访客，如果is_new标记是1，则对其状态进行修复。

import com.alibaba.fastjson.JSONObject;
import com.zhangbao.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
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.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;
import java.util.Date;

/*** @author: zhangbao* @date: 2021/6/18 23:29* @desc:**/
public class BaseLogTask {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//设置并行度，即kafka分区数env.setParallelism(4);//添加checkpoint，每5秒执行一次env.enableCheckpointing(5000, CheckpointingMode.EXACTLY\_ONCE);env.getCheckpointConfig().setCheckpointTimeout(60000);env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));//指定哪个用户读取hdfs文件System.setProperty("HADOOP\_USER\_NAME","zhangbao");//添加数据源，来至kafka的数据String topic = "ods\_base\_log";String group = "base\_log\_app\_group";FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, group);DataStreamSource<String> kafkaDs = env.addSource(kafkaSource);//对格式进行转换SingleOutputStreamOperator<JSONObject> jsonDs = kafkaDs.map(new MapFunction<String, JSONObject>() {@Overridepublic JSONObject map(String s) throws Exception {return JSONObject.parseObject(s);}});jsonDs.print("json >>> --- ");/*** 识别新老访客，前端会对新老客状态进行记录，可能不准，这里再次确认* 保存mid某天状态情况（将首次访问日期作为状态保存），等后面设备在有日志过来，从状态中获取日期和日志产生日期比较，* 如果状态不为空，并且状态日期和当前日期不相等，说明是老访客，如果is\_new标记是1，则对其状态进行修复*///根据id对日志进行分组KeyedStream<JSONObject, String> midKeyedDs = jsonDs.keyBy(data -> data.getJSONObject("common").getString("mid"));//新老访客状态修复，状态分为算子状态和键控状态，我们这里记录某一个设备状态，使用键控状态比较合适SingleOutputStreamOperator<JSONObject> midWithNewFlagDs = midKeyedDs.map(new RichMapFunction<JSONObject, JSONObject>() {//定义mid状态private ValueState<String> firstVisitDateState;//定义日期格式化private SimpleDateFormat sdf;//初始化方法@Overridepublic void open(Configuration parameters) throws Exception {firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("newMidDateState", String.class));sdf = new SimpleDateFormat("yyyyMMdd");}@Overridepublic JSONObject map(JSONObject jsonObject) throws Exception {//获取当前mid状态String is_new = jsonObject.getJSONObject("common").getString("is\_new");//获取当前日志时间戳Long ts = jsonObject.getLong("ts");if ("1".equals(is_new)) {//访客日期状态String stateDate = firstVisitDateState.value();String nowDate = sdf.format(new Date());if (stateDate != null && stateDate.length() != 0 && !stateDate.equals(nowDate)) {//是老客is_new = "0";jsonObject.getJSONObject("common").put("is\_new", is_new);} else {//新访客firstVisitDateState.update(nowDate);}}return jsonObject;}});
midWithNewFlagDs.print();try {//执行env.execute();} catch (Exception e) {e.printStackTrace();}}
}

5. 利用侧输出流实现数据拆分

经过上面的新老客户修复后，再将日志数据分为 3 类

启动日志标签定义：OutputTag startTag = new OutputTag("start"){};

和曝光日志标签定义：OutputTag displayTag = new OutputTag("display"){};

页面日志输出到主流，启动日志输出到启动侧输出流，曝光日志输出到曝光日志侧输出流。

数据拆分后发送到kafka

dwd_start_log：启动日志
dwd_display_log：曝光日志
dwd_page_log：页面日志

package com.zhangbao.gmall.realtime.app;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhangbao.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.text.SimpleDateFormat;
import java.util.Date;
/*** @author: zhangbao* @date: 2021/6/18 23:29* @desc:**/
public class BaseLogTask {private static final String TOPIC\_START = "dwd\_start\_log";private static final String TOPIC\_DISPLAY = "dwd\_display\_log";private static final String TOPIC\_PAGE = "dwd\_page\_log";public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//设置并行度，即kafka分区数env.setParallelism(4);//添加checkpoint，每5秒执行一次env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);env.getCheckpointConfig().setCheckpointTimeout(60000);env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));//指定哪个用户读取hdfs文件System.setProperty("HADOOP\_USER\_NAME","zhangbao");
//添加数据源，来至kafka的数据String topic = "ods\_base\_log";String group = "base\_log\_app\_group";FlinkKafkaConsumer kafkaSource = MyKafkaUtil.getKafkaSource(topic, group);DataStreamSource kafkaDs = env.addSource(kafkaSource);//对格式进行转换SingleOutputStreamOperator jsonDs = kafkaDs.map(new MapFunction() {@Overridepublic JSONObject map(String s) throws Exception {return JSONObject.parseObject(s);}});jsonDs.print("json >>> --- ");/*** 识别新老访客，前端会对新老客状态进行记录，可能不准，这里再次确认* 保存mid某天状态情况（将首次访问日期作为状态保存），等后面设备在有日志过来，从状态中获取日期和日志产生日期比较，* 如果状态不为空，并且状态日期和当前日期不相等，说明是老访客，如果is\_new标记是1，则对其状态进行修复*///根据id对日志进行分组KeyedStream midKeyedDs = jsonDs.keyBy(data -> data.getJSONObject("common").getString("mid"));//新老访客状态修复，状态分为算子状态和键控状态，我们这里记录某一个设备状态，使用键控状态比较合适SingleOutputStreamOperator midWithNewFlagDs = midKeyedDs.map(new RichMapFunction() {//定义mid状态private ValueState firstVisitDateState;//定义日期格式化private SimpleDateFormat sdf;//初始化方法@Overridepublic void open(Configuration parameters) throws Exception {firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor("newMidDateState", String.class));sdf = new SimpleDateFormat("yyyyMMdd");}@Overridepublic JSONObject map(JSONObject jsonObject) throws Exception {//获取当前mid状态String is\_new = jsonObject.getJSONObject("common").getString("is\_new");//获取当前日志时间戳Long ts = jsonObject.getLong("ts");if ("1".equals(is\_new)) {//访客日期状态String stateDate = firstVisitDateState.value();String nowDate = sdf.format(new Date());if (stateDate != null && stateDate.length() != 0 && !stateDate.equals(nowDate)) {//是老客is\_new = "0";jsonObject.getJSONObject("common").put("is\_new", is\_new);} else {//新访客firstVisitDateState.update(nowDate);}}return jsonObject;}});

//       midWithNewFlagDs.print();
/*** 根据日志数据内容,将日志数据分为 3 类, 页面日志、启动日志和曝光日志。页面日志* 输出到主流,启动日志输出到启动侧输出流,曝光日志输出到曝光日志侧输出流* 侧输出流：1接收迟到数据，2分流*///定义启动侧输出流标签，加大括号为了生成相应类型OutputTag startTag = new OutputTag("start"){};//定义曝光侧输出流标签OutputTag displayTag = new OutputTag("display"){};SingleOutputStreamOperator pageDs = midWithNewFlagDs.process(new ProcessFunction() {@Overridepublic void processElement(JSONObject jsonObject, Context context, Collector collector) throws Exception {String dataStr = jsonObject.toString();JSONObject startJson = jsonObject.getJSONObject("start");//判断是否启动日志if (startJson != null && startJson.size() > 0) {context.output(startTag, dataStr);} else {//判断是否曝光日志JSONArray jsonArray = jsonObject.getJSONArray("displays");if (jsonArray != null && jsonArray.size() > 0) {//给每一条曝光事件加pageIdString pageId = jsonObject.getJSONObject("page").getString("page\_id");//遍历输出曝光日志for (int i = 0; i < jsonArray.size(); i++) {JSONObject disPlayObj = jsonArray.getJSONObject(i);disPlayObj.put("page\_id", pageId);context.output(displayTag, disPlayObj.toString());}} else {//如果不是曝光日志，则是页面日志，输出到主流collector.collect(dataStr);}}}});
//获取侧输出流DataStream startDs = pageDs.getSideOutput(startTag);DataStream disPlayDs = pageDs.getSideOutput(displayTag);//打印输出startDs.print("start>>>");disPlayDs.print("display>>>");pageDs.print("page>>>");
/*** 将不同流的日志数据发送到指定的kafka主题*/startDs.addSink(MyKafkaUtil.getKafkaSink(TOPIC\_START));disPlayDs.addSink(MyKafkaUtil.getKafkaSink(TOPIC\_DISPLAY));pageDs.addSink(MyKafkaUtil.getKafkaSink(TOPIC\_PAGE));
try {//执行env.execute();} catch (Exception e) {e.printStackTrace();}}
}

4.Flink实时项目之数据拆分相关推荐

9.Flink实时项目之订单宽表
1.需求分析订单是统计分析的重要的对象,围绕订单有很多的维度统计需求,比如用户.地区.商品.品类.品牌等等.为了之后统计计算更加方便,减少大表之间的关联,所以在实时计算过程中将围绕订单的相关数据整合 ...
11.Flink实时项目之支付宽表
支付宽表支付宽表的目的,最主要的原因是支付表没有到订单明细,支付金额没有细分到商品上, 没有办法统计商品级的支付状况. 所以本次宽表的核心就是要把支付表的信息与订单明细关联上. 解决方案有两个一个 ...
FFA 2022 专场解读 - 实时风控实时湖仓数据集成
Flink Forward Asia 2022 将于 11 月 26-27 日在线上举办,议程内容正式上线! FFA 2022 官网:https://flink-forward.org.cn/ 实时风 ...
Flink实时计算指标对数方案
对于一个实时数据产品人员.或者开发人员来说,产品上展示的实时数据,pv.uv.gmv等等,怎么知道这些数据是不是正确的呢?当其他的小组开发的产品的数据(或者其他的数据提供方)又是另外一个数字,那么究竟 ...
flink实时流遇到的问题排查——部分数据未落库redis问题
flink实时流遇到的问题排查 1.技术和环境 2.问题表述 3.简化的代码 4.问题排查思路 5.结论 6.后续补充 1.技术和环境技术:kafka.zookeeper.DataStream.re ...
学习大数据-flink实时数据流处理
Flink得应用场景 IOT实时数据采集金融.银行得风控管理电商网站得实时数据报表什么是流处理.批处理流处理:数据是实时得批处理数据是等待达到某一批后再处理数据 DataStream API ...
实时可视化大数据项目02 -- 项目目录介绍
1.项目目录结构后端采用分层架构,前端springBoot整合了前端. 2.目录介绍 2.1.dao 存放的是实体类,基本是封装好的对象,传递给前端的对象. 2.2.service 业务层: 封装数 ...
Flink系列之：基于scala语言实现flink实时消费Kafka Topic中的数据
Flink系列之:基于scala语言实现flink实时消费Kafka Topic中的数据一.引入flink相关依赖二.properties保存连接kafka的配置三.构建flink实时消费环境 ...
Flink SQL 实时大屏(实时查询存量数据-批转流)
最近接到一个需求,关于flink实时大屏需求.每半小时展示历史每天当前半小时(每天00:00:00-00:30:00之间) 的数据的最大值.最小值.中位数.上四分位数.下四分位数. 需求描述每半小时 ...

4.Flink实时项目之数据拆分

Python微信订餐小程序课程视频

Python实战量化交易理财系统