用户行为日志DWD层实现目标：

1. 识别新老用户，虽然客户端有新老用户的标识，但是不准确，需要用实时计算再次确认
2. 利用侧输出流实现数据拆分，根据日志数据内容,将日志数据分为3类, 页面日志、启动日志和曝光日志。页面日志输出到主流,启动日志输出到启动侧输出流,曝光日志输出到曝光日志侧输出流
3. 将不同流的数据推送下游的 Kafka 的不同 Topic 中

代码实现：

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.MyKafkaUtil;
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.streaming.api.datastream.DataStream;
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.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;public class BaseLogApp {public static void main(String[] args) throws Exception {//TODO 1.获取执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//1.1 设置CK&状态后端//env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall-flink-210325/ck"));//env.enableCheckpointing(5000L);//env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//env.getCheckpointConfig().setCheckpointTimeout(10000L);//env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);//env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);//env.setRestartStrategy(RestartStrategies.fixedDelayRestart());//TODO 2.消费 ods_base_log 主题数据创建流String sourceTopic = "ods_base_log";String groupId = "base_log_app_210325";DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));//TODO 3.将每行数据转换为JSON对象OutputTag<String> outputTag = new OutputTag<String>("Dirty") {};SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {@Overridepublic void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {try {JSONObject jsonObject = JSON.parseObject(value);out.collect(jsonObject);} catch (Exception e) {//发生异常,将数据写入侧输出流ctx.output(outputTag, value);}}});//打印脏数据jsonObjDS.getSideOutput(outputTag).print("Dirty>>>>>>>>>>>");//TODO 4.新老用户校验  状态编程SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlagDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid")).map(new RichMapFunction<JSONObject, JSONObject>() {private ValueState<String> valueState;@Overridepublic void open(Configuration parameters) throws Exception {valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", String.class));}@Overridepublic JSONObject map(JSONObject value) throws Exception {//获取数据中的"is_new"标记String isNew = value.getJSONObject("common").getString("is_new");//判断isNew标记是否为"1"if ("1".equals(isNew)) {//获取状态数据String state = valueState.value();if (state != null) {//修改isNew标记value.getJSONObject("common").put("is_new", "0");} else {valueState.update("1");}}return value;}});//TODO 5.分流  侧输出流  页面：主流  启动：侧输出流  曝光：侧输出流OutputTag<String> startTag = new OutputTag<String>("start") {};OutputTag<String> displayTag = new OutputTag<String>("display") {};SingleOutputStreamOperator<String> pageDS = jsonObjWithNewFlagDS.process(new ProcessFunction<JSONObject, String>() {@Overridepublic void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {//获取启动日志字段String start = value.getString("start");if (start != null && start.length() > 0) {//将数据写入启动日志侧输出流ctx.output(startTag, value.toJSONString());} else {//将数据写入页面日志主流out.collect(value.toJSONString());//取出数据中的曝光数据JSONArray displays = value.getJSONArray("displays");if (displays != null && displays.size() > 0) {//获取页面IDString pageId = value.getJSONObject("page").getString("page_id");for (int i = 0; i < displays.size(); i++) {JSONObject display = displays.getJSONObject(i);//添加页面iddisplay.put("page_id", pageId);//将输出写出到曝光侧输出流ctx.output(displayTag, display.toJSONString());}}}}});//TODO 6.提取侧输出流DataStream<String> startDS = pageDS.getSideOutput(startTag);DataStream<String> displayDS = pageDS.getSideOutput(displayTag);//TODO 7.将三个流进行打印并输出到对应的Kafka主题中startDS.print("Start>>>>>>>>>>>");pageDS.print("Page>>>>>>>>>>>");displayDS.print("Display>>>>>>>>>>>>");startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_start_log"));pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_page_log"));displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_display_log"));//TODO 8.启动任务env.execute("BaseLogApp");}}