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

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>() {
            @Override
            public 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>() {
            @Override
            public 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;
            //初始化方法
            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("newMidDateState", String.class));
                sdf = new SimpleDateFormat("yyyyMMdd");
            }
            @Override
            public 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<String> startTag = new OutputTag<String>("start"){};

和曝光日志标签定义：OutputTag<String> displayTag = new OutputTag<String>("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<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDs = env.addSource(kafkaSource);
        //对格式进行转换
        SingleOutputStreamOperator<JSONObject> jsonDs = kafkaDs.map(new MapFunction<String, JSONObject>() {
            @Override
            public 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;
            //初始化方法
            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("newMidDateState", String.class));
                sdf = new SimpleDateFormat("yyyyMMdd");
            }
            @Override
            public 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<String> startTag = new OutputTag<String>("start"){};
        //定义曝光侧输出流标签
        OutputTag<String> displayTag = new OutputTag<String>("display"){};
        SingleOutputStreamOperator<String> pageDs = midWithNewFlagDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<String> 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) {
                                //给每一条曝光事件加pageId
                                String 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<String> startDs = pageDs.getSideOutput(startTag);
        DataStream<String> 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();
        }
    }
}

最后编辑于：2022.02.19 20:32:53

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 204,445评论 6赞 478
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 85,889评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 151,047评论 0赞 337
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,760评论 1赞 276
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,745评论 5赞 367
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,638评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,011评论 3赞 398
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,669评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 40,923评论 1赞 299
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,655评论 2赞 321
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,740评论 1赞 330
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,406评论 4赞 320
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 38,995评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,961评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,197评论 1赞 260
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,023评论 2赞 350
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,483评论 2赞 342