当前位置：网站首页>4. Data splitting of Flink real-time project

4. Data splitting of Flink real-time project

2022-07-03 21:27:00 【Contestant No.1 position】

1. Abstract

The log data we collected earlier has been saved to Kafka in , As log data ODS layer , from kafka Of ODS The log data read by the layer is divided into 3 class , Page log 、 Start log and exposure log . Although these three types of data are user behavior data , But it has a completely different data structure , So we need to split it . Write the split logs back to Kafka In different topics , As log DWD layer .

The page log is output to the main stream , The startup log is output to the startup side output stream , The exposure log is output to the exposure side output stream

2. Identify new and old users

The client business itself has the identification of new and old users , But it's not accurate enough , Need to reconfirm with real-time calculation ( Business operations are not involved , Just make a simple status confirmation ).

Data splitting is realized by using side output stream

According to the content of log data , Divide the log data into 3 class ： Page log 、 Start log and exposure log . Push the data of different streams to the downstream kafka Different Topic in

3. Code implementation

In the bag app Create flink Mission BaseLogTask.java,

adopt flink consumption kafka The data of , Then record the consumption checkpoint Deposit in hdfs in , Remember to create the path manually , Then give the authority

checkpoint Optional use , You can turn it off during the test .

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();
        // Set parallelism , namely kafka Partition number 
        env.setParallelism(4);
        // add to checkpoint, Every time 5 Once per second 
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));
        // Specify which user reads hdfs file 
        System.setProperty("HADOOP_USER_NAME","zhangbao");
        // Add data sources 
        String topic = "ods_base_log";
        String group = "base_log_app_group";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDs = env.addSource(kafkaSource);
        // Convert the format 
        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 {
            // perform 
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.

MyKafkaUtil.java Tool class

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 consumer  */
    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);
    }
}

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.

4. New and old visitor status repair

Rules for identifying new and old customers

Identify new and old visitors , The front end will record the status of new and old customers , Maybe not , Here again , preservation mid State of one day （ Save first visit date as status ）, Wait until there is a log in the back equipment , Get the date from the status and compare the log generation date , If the status is not empty , And the status date is not equal to the current date , It means it's a regular visitor , If is_new Mark is 1, Then repair its state .

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();
        // Set parallelism , namely kafka Partition number 
        env.setParallelism(4);
        // add to checkpoint, Every time 5 Once per second 
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));
        // Specify which user reads hdfs file 
        System.setProperty("HADOOP_USER_NAME","zhangbao");
        
        // Add data sources , Come here kafka The data of 
        String topic = "ods_base_log";
        String group = "base_log_app_group";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDs = env.addSource(kafkaSource);
        // Convert the format 
        SingleOutputStreamOperator<JSONObject> jsonDs = kafkaDs.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSONObject.parseObject(s);
            }
        });
        jsonDs.print("json >>> --- ");
        /** *  Identify new and old visitors , The front end will record the status of new and old customers , Maybe not , Here again  *  preservation mid State of one day （ Save first visit date as status ）, Wait until there is a log in the back equipment , Get the date from the status and compare the log generation date , *  If the status is not empty , And the status date is not equal to the current date , It means it's a regular visitor , If is_new Mark is 1, Then repair its state  */
        // according to id Group logs 
        KeyedStream<JSONObject, String> midKeyedDs = jsonDs.keyBy(data -> data.getJSONObject("common").getString("mid"));
        // New and old visitor status repair , States are divided into operator States and keyed states , Here we record the status of a certain device , Using keyed state is more appropriate 
        SingleOutputStreamOperator<JSONObject> midWithNewFlagDs = midKeyedDs.map(new RichMapFunction<JSONObject, JSONObject>() {
            // Definition mid state 
            private ValueState<String> firstVisitDateState;
            // Define date formatting 
            private SimpleDateFormat sdf;
            // Initialization method 
            @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 {
                // Get current mid state 
                String is_new = jsonObject.getJSONObject("common").getString("is_new");
                // Get the current log timestamp 
                Long ts = jsonObject.getLong("ts");
                if ("1".equals(is_new)) {
                    // Visitor date status 
                    String stateDate = firstVisitDateState.value();
                    String nowDate = sdf.format(new Date());
                    if (stateDate != null && stateDate.length() != 0 && !stateDate.equals(nowDate)) {
                        // It's an old guest 
                        is_new = "0";
                        jsonObject.getJSONObject("common").put("is_new", is_new);
                    } else {
                        // New visitors 
                        firstVisitDateState.update(nowDate);
                    }
                }
                return jsonObject;
            }
        });

        midWithNewFlagDs.print();
        try {
            // perform 
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.

5. Data splitting is realized by using side output stream

After the above new and old customers repair , Then divide the log data into 3 class

Start log label definition ： OutputTag<String> startTag = new OutputTag<String>("start"){};

And exposure log label definitions ： OutputTag<String> displayTag = new OutputTag<String>("display"){};

The page log is output to the main stream , The startup log is output to the startup side output stream , The exposure log is output to the exposure log side output stream .

The data is split and sent to kafka

dwd_start_log： start log
dwd_display_log： Exposure log
dwd_page_log： 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();
        // Set parallelism , namely kafka Partition number 
        env.setParallelism(4);
        // add to checkpoint, Every time 5 Once per second 
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9000/gmall/flink/checkpoint/baseLogAll"));
        // Specify which user reads hdfs file 
        System.setProperty("HADOOP_USER_NAME","zhangbao");

        // Add data sources , Come here kafka The data of 
        String topic = "ods_base_log";
        String group = "base_log_app_group";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDs = env.addSource(kafkaSource);
        // Convert the format 
        SingleOutputStreamOperator<JSONObject> jsonDs = kafkaDs.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSONObject.parseObject(s);
            }
        });
        jsonDs.print("json >>> --- ");
        /** *  Identify new and old visitors , The front end will record the status of new and old customers , Maybe not , Here again  *  preservation mid State of one day （ Save first visit date as status ）, Wait until there is a log in the back equipment , Get the date from the status and compare the log generation date , *  If the status is not empty , And the status date is not equal to the current date , It means it's a regular visitor , If is_new Mark is 1, Then repair its state  */
        // according to id Group logs 
        KeyedStream<JSONObject, String> midKeyedDs = jsonDs.keyBy(data -> data.getJSONObject("common").getString("mid"));
        // New and old visitor status repair , States are divided into operator States and keyed states , Here we record the status of a certain device , Using keyed state is more appropriate 
        SingleOutputStreamOperator<JSONObject> midWithNewFlagDs = midKeyedDs.map(new RichMapFunction<JSONObject, JSONObject>() {
            // Definition mid state 
            private ValueState<String> firstVisitDateState;
            // Define date formatting 
            private SimpleDateFormat sdf;
            // Initialization method 
            @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 {
                // Get current mid state 
                String is_new = jsonObject.getJSONObject("common").getString("is_new");
                // Get the current log timestamp 
                Long ts = jsonObject.getLong("ts");
                if ("1".equals(is_new)) {
                    // Visitor date status 
                    String stateDate = firstVisitDateState.value();
                    String nowDate = sdf.format(new Date());
                    if (stateDate != null && stateDate.length() != 0 && !stateDate.equals(nowDate)) {
                        // It's an old guest 
                        is_new = "0";
                        jsonObject.getJSONObject("common").put("is_new", is_new);
                    } else {
                        // New visitors 
                        firstVisitDateState.update(nowDate);
                    }
                }
                return jsonObject;
            }
        });

// midWithNewFlagDs.print();

        /** *  According to the content of log data , Divide the log data into  3  class ,  Page log 、 Start log and exposure log . Page log  *  Output to the mainstream , The startup log is output to the startup side output stream , The exposure log is output to the exposure log side output stream  *  Side output stream ：1 Receive late data ,2 shunt  */
        // Define the label of the output stream on the startup side , Open parentheses to generate the corresponding type 
        OutputTag<String> startTag = new OutputTag<String>("start"){};
        // Define the exposure side output stream label 
        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");
                        // Determine whether to start the log 
                        if (startJson != null && startJson.size() > 0) {
                            context.output(startTag, dataStr);
                        } else {
                            // Determine whether to expose the log 
                            JSONArray jsonArray = jsonObject.getJSONArray("displays");
                            if (jsonArray != null && jsonArray.size() > 0) {
                                // Add... To each exposure pageId
                                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                                // Traverse the output exposure log 
                                for (int i = 0; i < jsonArray.size(); i++) {
                                    JSONObject disPlayObj = jsonArray.getJSONObject(i);
                                    disPlayObj.put("page_id", pageId);
                                    context.output(displayTag, disPlayObj.toString());
                                }
                            } else {
                                // If it's not the exposure log , Page log , Output to the mainstream 
                                collector.collect(dataStr);
                            }
                        }
                    }
                }
        );

        // Get side output stream 
        DataStream<String> startDs = pageDs.getSideOutput(startTag);
        DataStream<String> disPlayDs = pageDs.getSideOutput(displayTag);
        // Printout 
        startDs.print("start>>>");
        disPlayDs.print("display>>>");
        pageDs.print("page>>>");

        /** *  Send the log data of different streams to the specified kafka The theme  */
        startDs.addSink(MyKafkaUtil.getKafkaSink(TOPIC_START));
        disPlayDs.addSink(MyKafkaUtil.getKafkaSink(TOPIC_DISPLAY));
        pageDs.addSink(MyKafkaUtil.getKafkaSink(TOPIC_PAGE));

        try {
            // perform 
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.
159.
160.
161.