飙血推荐
  • HTML教程
  • MySQL教程
  • JavaScript基础教程
  • php入门教程
  • JavaScript正则表达式运用
  • Excel函数教程
  • UEditor使用文档
  • AngularJS教程
  • ThinkPHP5.0教程

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

时间：2022-02-15  作者：zhangbaohpu  

1. 摘要

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

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

2. 识别新老用户

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

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

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

3. 代码实现

在包app下创建flink任务域名，

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

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

package 域名域名;
import 域名域名Object;
import 域名域名域名fkaUtil;
import 域名域名域名unction;
import 域名域名域名ateBackend;
import 域名域名.CheckpointingMode;
import 域名域名.域名StreamSource;
import 域名域名.域名leOutputStreamOperator;
import 域名域名.域名amExecutionEnvironment;
import 域名域名域名kKafkaConsumer;
/**
 * @author: zhangbao
 * @date: 2021/6/18 23:29
 * @desc:
 **/
public class BaseLogTask {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = 域名xecutionEnvironment();
        //设置并行度，即kafka分区数
        域名arallelism(4);
        //添加checkpoint，每5秒执行一次
        域名leCheckpointing(5000, 域名TLY_ONCE);
        域名heckpointConfig().setCheckpointTimeout(60000);
        域名tateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));
        //指定哪个用户读取hdfs文件
        域名roperty("HADOOP_USER_NAME","zhangbao");
        //添加数据源
        String topic = "ods_base_log";
        String group = "base_log_app_group";
        FlinkKafkaConsumer<String> kafkaSource = 域名afkaSource(topic, group);
        DataStreamSource<String> kafkaDs = 域名ource(kafkaSource);
        //对格式进行转换
        SingleOutputStreamOperator<JSONObject> jsonDs = 域名(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return 域名eObject(s);
            }
        });
        域名t("json >>> --- ");
​
        try {
            //执行
            域名ute();
        } catch (Exception e) {
            域名tStackTrace();
        }
​
    }
}

域名工具类

package 域名域名s;
import 域名域名域名leStringSchema;
import 域名域名域名kKafkaConsumer;
import 域名域名域名umerConfig;
import 域名erties;
/**
 * @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();
        域名roperty(域名P_ID_CONFIG,group);
        域名roperty(域名STRAP_SERVERS_CONFIG,kafka_host);
        return new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(),props);
    }
}

4. 新老访客状态修复

识别新老客户规则

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

import 域名域名Object;
import 域名域名域名fkaUtil;
import 域名域名域名unction;
import 域名域名域名MapFunction;
import 域名域名域名eState;
import 域名域名域名eStateDescriptor;
import 域名域名iguration;
import 域名域名域名ateBackend;
import 域名域名.CheckpointingMode;
import 域名域名.域名StreamSource;
import 域名域名.域名dStream;
import 域名域名.域名leOutputStreamOperator;
import 域名域名.域名amExecutionEnvironment;
import 域名域名域名kKafkaConsumer;
​
import 域名leDateFormat;
import 域名;
​
/**
 * @author: zhangbao
 * @date: 2021/6/18 23:29
 * @desc:
 **/
public class BaseLogTask {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = 域名xecutionEnvironment();
        //设置并行度，即kafka分区数
        域名arallelism(4);
        //添加checkpoint，每5秒执行一次
        域名leCheckpointing(5000, 域名TLY_ONCE);
        域名heckpointConfig().setCheckpointTimeout(60000);
        域名tateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));
        //指定哪个用户读取hdfs文件
        域名roperty("HADOOP_USER_NAME","zhangbao");
        
        //添加数据源，来至kafka的数据
        String topic = "ods_base_log";
        String group = "base_log_app_group";
        FlinkKafkaConsumer<String> kafkaSource = 域名afkaSource(topic, group);
        DataStreamSource<String> kafkaDs = 域名ource(kafkaSource);
        //对格式进行转换
        SingleOutputStreamOperator<JSONObject> jsonDs = 域名(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return 域名eObject(s);
            }
        });
        域名t("json >>> --- ");
        /**
         * 识别新老访客，前端会对新老客状态进行记录，可能不准，这里再次确认
         * 保存mid某天状态情况（将首次访问日期作为状态保存），等后面设备在有日志过来，从状态中获取日期和日志产生日期比较，
         * 如果状态不为空，并且状态日期和当前日期不相等，说明是老访客，如果is_new标记是1，则对其状态进行修复
         */
        //根据id对日志进行分组
        KeyedStream<JSONObject, String> midKeyedDs = 域名y(data -> 域名SONObject("common").getString("mid"));
        //新老访客状态修复，状态分为算子状态和键控状态，我们这里记录某一个设备状态，使用键控状态比较合适
        SingleOutputStreamOperator<JSONObject> midWithNewFlagDs = 域名(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", 域名s));
                sdf = new SimpleDateFormat("yyyyMMdd");
            }
            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                //获取当前mid状态
                String is_new = 域名SONObject("common").getString("is_new");
                //获取当前日志时间戳
                Long ts = 域名ong("ts");
                if ("1".equals(is_new)) {
                    //访客日期状态
                    String stateDate = 域名e();
                    String nowDate = 域名at(new Date());
                    if (stateDate != null && 域名th() != 0 && !域名ls(nowDate)) {
                        //是老客
                        is_new = "0";
                        域名SONObject("common").put("is_new", is_new);
                    } else {
                        //新访客
                        域名te(nowDate);
                    }
                }
                return jsonObject;
            }
        });
​
        域名t();
        try {
            //执行
            域名ute();
        } catch (Exception e) {
            域名tStackTrace();
        }
    }
}

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 域名域名;
import 域名域名Array;
import 域名域名Object;
import 域名域名域名fkaUtil;
import 域名域名域名unction;
import 域名域名域名MapFunction;
import 域名域名域名eState;
import 域名域名域名eStateDescriptor;
import 域名域名iguration;
import 域名域名域名ateBackend;
import 域名域名.CheckpointingMode;
import 域名域名.域名Stream;
import 域名域名.域名StreamSource;
import 域名域名.域名dStream;
import 域名域名.域名leOutputStreamOperator;
import 域名域名.域名amExecutionEnvironment;
import 域名域名.域名essFunction;
import 域名域名域名kKafkaConsumer;
import 域名域名ector;
import 域名域名utTag;
import 域名leDateFormat;
import 域名;
/**
 * @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 = 域名xecutionEnvironment();
        //设置并行度，即kafka分区数
        域名arallelism(4);
        //添加checkpoint，每5秒执行一次
        域名leCheckpointing(5000, 域名TLY_ONCE);
        域名heckpointConfig().setCheckpointTimeout(60000);
        域名tateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));
        //指定哪个用户读取hdfs文件
        域名roperty("HADOOP_USER_NAME","zhangbao");
​
        //添加数据源，来至kafka的数据
        String topic = "ods_base_log";
        String group = "base_log_app_group";
        FlinkKafkaConsumer<String> kafkaSource = 域名afkaSource(topic, group);
        DataStreamSource<String> kafkaDs = 域名ource(kafkaSource);
        //对格式进行转换
        SingleOutputStreamOperator<JSONObject> jsonDs = 域名(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return 域名eObject(s);
            }
        });
        域名t("json >>> --- ");
        /**
         * 识别新老访客，前端会对新老客状态进行记录，可能不准，这里再次确认
         * 保存mid某天状态情况（将首次访问日期作为状态保存），等后面设备在有日志过来，从状态中获取日期和日志产生日期比较，
         * 如果状态不为空，并且状态日期和当前日期不相等，说明是老访客，如果is_new标记是1，则对其状态进行修复
         */
        //根据id对日志进行分组
        KeyedStream<JSONObject, String> midKeyedDs = 域名y(data -> 域名SONObject("common").getString("mid"));
        //新老访客状态修复，状态分为算子状态和键控状态，我们这里记录某一个设备状态，使用键控状态比较合适
        SingleOutputStreamOperator<JSONObject> midWithNewFlagDs = 域名(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", 域名s));
                sdf = new SimpleDateFormat("yyyyMMdd");
            }
            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                //获取当前mid状态
                String is_new = 域名SONObject("common").getString("is_new");
                //获取当前日志时间戳
                Long ts = 域名ong("ts");
                if ("1".equals(is_new)) {
                    //访客日期状态
                    String stateDate = 域名e();
                    String nowDate = 域名at(new Date());
                    if (stateDate != null && 域名th() != 0 && !域名ls(nowDate)) {
                        //是老客
                        is_new = "0";
                        域名SONObject("common").put("is_new", is_new);
                    } else {
                        //新访客
                        域名te(nowDate);
                    }
                }
                return jsonObject;
            }
        });
​
//        域名t();
​
        /**
         * 根据日志数据内容,将日志数据分为 3 类, 页面日志、启动日志和曝光日志。页面日志
         * 输出到主流,启动日志输出到启动侧输出流,曝光日志输出到曝光日志侧输出流
         * 侧输出流：1接收迟到数据，2分流
         */
        //定义启动侧输出流标签，加大括号为了生成相应类型
        OutputTag<String> startTag = new OutputTag<String>("start"){};
        //定义曝光侧输出流标签
        OutputTag<String> displayTag = new OutputTag<String>("display"){};
        SingleOutputStreamOperator<String> pageDs = 域名ess(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {
                        String dataStr = 域名ring();
                        JSONObject startJson = 域名SONObject("start");
                        //判断是否启动日志
                        if (startJson != null && 域名() > 0) {
                            域名ut(startTag, dataStr);
                        } else {
                            //判断是否曝光日志
                            JSONArray jsonArray = 域名SONArray("displays");
                            if (jsonArray != null && 域名() > 0) {
                                //给每一条曝光事件加pageId
                                String pageId = 域名SONObject("page").getString("page_id");
                                //遍历输出曝光日志
                                for (int i = 0; i < 域名(); i++) {
                                    JSONObject disPlayObj = 域名SONObject(i);
                                    域名("page_id", pageId);
                                    域名ut(displayTag, 域名ring());
                                }
                            } else {
                                //如果不是曝光日志，则是页面日志，输出到主流
                                域名ect(dataStr);
                            }
                        }
                    }
                }
        );
​
        //获取侧输出流
        DataStream<String> startDs = 域名ideOutput(startTag);
        DataStream<String> disPlayDs = 域名ideOutput(displayTag);
        //打印输出
        域名t("start>>>");
        域名t("display>>>");
        域名t("page>>>");
​
        /**
         * 将不同流的日志数据发送到指定的kafka主题
         */
        域名ink(域名afkaSink(TOPIC_START));
        域名ink(域名afkaSink(TOPIC_DISPLAY));
        域名ink(域名afkaSink(TOPIC_PAGE));
​
        try {
            //执行
            域名ute();
        } catch (Exception e) {
            域名tStackTrace();
        }
    }
}

标签：编程
搜你所爱
大家在看
热搜标签
网站地图 | 注册 | 投稿
白菜窝
湘ICP备14001474号-3  投诉建议:234161800@qq.com   部分内容来源于网络，如有侵权，请联系删除。