一个简单的Flink程序开发代码

青旅半醒 2021-12-04 00:57 267阅读 0赞

>     重点主要是使用State，模拟实现匹配场景功能 ，想了解更多完整代码，可以留言
>     
>     package com.coder.flink.core.stormToFlink;
>     
>     
>     import com.alibaba.fastjson.JSON;
>     import com.alibaba.fastjson.JSONArray;
>     import com.alibaba.fastjson.JSONObject;
>     import com.alibaba.fastjson.TypeReference;
>     import com.coder.flink.core.base.Constaints;
>     import com.coder.flink.core.base.Resources;
>     import com.coder.flink.core.pojo.TopicInfo;
>     import com.coder.flink.core.quartz.QuartzRedis;
>     import com.coder.flink.core.redis.RedisTableData;
>     import com.coder.flink.core.utils.DateUtil;
>     import com.coder.flink.core.utils.GsonUtil;
>     import com.coder.flink.core.utils.HbaseRowKey;
>     import com.coder.flink.core.utils.HbaseUtil;
>     import com.google.gson.JsonArray;
>     import com.google.gson.JsonObject;
>     import com.googlecode.aviator.Expression;
>     import org.apache.commons.lang.StringUtils;
>     import org.apache.flink.api.common.functions.*;
>     import org.apache.flink.api.common.serialization.SimpleStringSchema;
>     import org.apache.flink.api.common.state.MapState;
>     import org.apache.flink.api.common.state.MapStateDescriptor;
>     import org.apache.flink.api.java.functions.KeySelector;
>     import org.apache.flink.api.java.tuple.Tuple2;
>     import org.apache.flink.api.java.utils.ParameterTool;
>     import org.apache.flink.configuration.Configuration;
>     import org.apache.flink.streaming.api.datastream.DataStream;
>     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.source.RichSourceFunction;
>     import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
>     import org.apache.flink.streaming.api.windowing.time.Time;
>     import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
>     import org.apache.flink.util.Collector;
>     import org.apache.hadoop.hbase.Cell;
>     import org.apache.hadoop.hbase.CellUtil;
>     import org.apache.hadoop.hbase.client.Get;
>     import org.apache.hadoop.hbase.client.Put;
>     import org.apache.hadoop.hbase.client.Result;
>     import org.apache.hadoop.hbase.util.Bytes;
>     import org.apache.kafka.clients.consumer.ConsumerConfig;
>     
>     import java.util.*;
>     import java.util.concurrent.ConcurrentHashMap;
>     import java.util.concurrent.TimeUnit;
>     
>     
>     /**
>      * 最后需要把方法封装
>      */
>     public class StormToFlink_new {
>     
>         private static String columnFamily = Resources.hbasePro.getProperty("column_family");
>     
>         public static void main(String[] args) throws Exception {
>     
>             //todo 1， 启动定时器
>             new QuartzRedis();
>     
>             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>     
>             // todo 2，代码模拟JSON 日志
>     
>             String log1 = "{'rowKey':'002|AA|4BCED55E25123F5A','phoneType':'1','city':'深圳市','urlTime':'[21/Jun/2019:21:23:42 +0800]','latitude':'22.495816','battery':'34','deviceId':'E4A787938C2EBFBD0','memoryUseRatio':'41','localTime':'1558337193164','areaCode':'42287','baseStationCode':'23082003','province':'广东省','interfaceKey':'Tub','currentIPAddress':'172.10.13.55','fieldUrl':'/publicLog/domainName','appPackageName':'com.liumeng.demo','cpuUseRatio':'69','appKey':'432b6d2931efd34037fe282d'}";
>             String log2 = "{'rowKey':'002|AA|4BCED55E25123F5A','phoneType':'1','city':'深圳市','urlTime':'[21/Jun/2019:21:23:48 +0800]','latitude':'22.495816','battery':'34','deviceId':'E4A787938C2EBFBD0','memoryUseRatio':'41','localTime':'1558337193381','areaCode':'42287','baseStationCode':'23082003','province':'广东省','interfaceKey':'YRC','currentIPAddress':'172.10.13.55','fieldUrl':'/publicLog/sensitivEspec','appPackageName':'com.liumeng.demo','cpuUseRatio':'58','appKey':'432b6d2931efd34037fe282d',}";
>             String log3 = "{'rowKey':'002|AA|4BCED55E25123F5A','phoneType':'1','city':'深圳市','urlTime':'[21/Jun/2019:21:24:10 +0800]','latitude':'22.495816','battery':'34','deviceId':'E4A787938C2EBFBD0','memoryUseRatio':'41','localTime':'1558337193189','areaCode':'42287','baseStationCode':'23082003','province':'广东省','interfaceKey':'xhg','currentIPAddress':'172.10.13.55','fieldUrl':'/publicLog/libc','appKey':'432b6d2931efd34037fe282d','shortAppKey':'AA'}";
>             String log4 = "{'rowKey':'002|AA|4BCED55E25123F5A','city':'北京市','urlTime':'[21/Jun/2019:21:24:19 +0800]','virusAppLevel':4,'deviceId':'58553C9A0974CFDB0','province':'北京','appKey':'432b6d2931efd34037fe282d','virusAppPath':'/data/app/com.Copon-1/base.apk0','longitude':'116.397128','phoneType':'1','virusAppMd':'74e09c5f57d5a040c86a86cdad7f04fa','isStubbornVirusApp':false,'interfaceKey':'AiV','fieldUrl':'/publicLog/appVirus'}";
>             String log5 = "{'rowKey':'002|AA|4BCED55E25123F5A','urlTime':'[21/Jun/2019:21:24:27 +0800]','deviceId':'E4A787938C2EBFBD0','isFirstVisit':true,'baseStationCode':'23082003','province':'北京','appKey':'432b6d2931efd34037fe282d','networkType':'wifi','longitude':'116.397128','phoneType':'1','interfaceKey':'AiT','fieldUrl':'/publicLog/appPage','appPackageName':'com.liumeng.demo','shortAppKey':'AA'}";
>     
>             List<String> logStrList = new ArrayList<>();
>             logStrList.add(log1);
>             logStrList.add(log2);
>             logStrList.add(log3);
>             logStrList.add(log4);
>             logStrList.add(log5);
>     
>     
>             DataStream<String> dataStream = env.addSource(new RichSourceFunction<String>() {
>                 private volatile boolean isRunning = true;
>     
>                 @Override
>                 public void run(SourceContext<String> ctx) throws Exception {
>                     while (isRunning) {
>                         Random random = new Random();
>                         TimeUnit.SECONDS.sleep(3);
>                         int i = random.nextInt(5);
>                         ctx.collect(logStrList.get(i));
>                     }
>                 }
>     
>                 @Override
>                 public void cancel() {
>                     isRunning = false;
>                 }
>             }).setParallelism(1);
>     
>             //todo 3，过滤掉不满足格式的数据
>     
>             DataStream<JSONObject> jsonDstream = dataStream.filter(new FilterFunction<String>() {
>                 @Override
>                 public boolean filter(String value) {
>                     try {
>                         JSONObject filterJson = JSON.parseObject(value);
>                         return true;
>                     } catch (Exception ex) {
>                         System.out.println("此日志不符合规范");
>                     }
>                     return false;
>                 }
>             }).uid("filter_logs").map(new MapFunction<String, JSONObject>() {
>                 @Override
>                 public JSONObject map(String value) {
>                     Map<String, List<TopicInfo>> returnMap = RedisTableData.returnMap;
>                     JSONObject logJson = JSON.parseObject(value);
>                     return logJson;
>                 }
>             }).uid("map_logs");
>     
>     
>             //todo 4，查询hbase的数据，补全日志
>             DataStream<JSONObject> queryHbaseData = jsonDstream.map(new RichMapFunction<JSONObject, JSONObject>() {
>     
>                 @Override
>                 public void open(Configuration parameters) throws Exception {
>                     super.open(parameters);
>                     HbaseUtil.getInstance();
>                 }
>     
>                 @Override
>                 public JSONObject map(JSONObject logJson) throws Exception {
>                     JSONObject getDataLog = StormUtil.queryHbase(logJson);
>                     //todo 转成Map
>     
>                     return getDataLog;
>                 }
>             }).uid("query_hbase");
>     
>     
>             //todo 5，按照key分组，然后匹配第一个规则
>             DataStream<JSONObject> topicNameDstream = queryHbaseData.keyBy(new KeySelector<JSONObject, String>() {
>                 @Override
>                 public String getKey(JSONObject logJson) throws Exception {
>                     String logAppKey = logJson.getString("appKey");
>                     return logAppKey;
>                 }
>             }).flatMap(new RichFlatMapFunction<JSONObject, JSONObject>() {
>                 private Map<String, List<TopicInfo>> topicInfosMap;
>                 private Map<String, Expression> logExpressionMap;
>                 private Map<String, Expression> sequenceExpressionMap;
>     
>                 //todo 状态为True 或者 False
>                 private transient MapState<String, String> expressionState;
>     
>                 @Override
>                 public void open(Configuration parameters) throws Exception {
>                     super.open(parameters);
>                     topicInfosMap = RedisTableData.returnMap;
>                     logExpressionMap = RedisTableData.mapExpression;
>                     sequenceExpressionMap = RedisTableData.boltTransferMap;
>     
>                     expressionState = getRuntimeContext().getMapState(
>                             new MapStateDescriptor<String, String>(
>                                     "expressionState",
>                                     String.class,
>                                     String.class));
>                 }
>     
>                 @Override
>                 public void flatMap(JSONObject logJson, Collector<JSONObject> out) throws Exception {
>     
>                     System.out.println("topicInfosMap.size() = " + topicInfosMap.size());
>                     Map<String, Object> logCompareMap = JSON.parseObject(logJson.toString(), new TypeReference<Map<String, Object>>() {
>                     });
>     
>                     String logAppKey = logJson.getString("appKey");
>                     String interfaceKey = logJson.getString("interfaceKey");
>     
>                     String findKey = logAppKey + ":" + interfaceKey;
>                     // 预处理包含这个日志key
>                     if (topicInfosMap.containsKey(findKey)) {
>                         List<TopicInfo> topicInfoList = topicInfosMap.get(findKey);
>     
>                         if (topicInfoList != null) {
>     
>                             for (int i = 0; i < topicInfoList.size(); i++) {
>     
>                                 TopicInfo topicInfoBean = topicInfoList.get(i);
>                                 String topicName = topicInfoBean.getTopicName();
>                                 List<String> interFaceKeyList = topicInfoBean.getInterFaceKeyList();
>     
>                                 //todo 将查询的数据插入到这个Map里面去，然后调用引擎表达式
>                                 Map typeExpressionMap = new HashMap<String, Object>();
>     
>                                 //  匹配单个场景表达式的多个条件
>                                 for (int j = 0; j < interFaceKeyList.size(); j++) {
>                                     String everyKey = interFaceKeyList.get(j);
>                                     String completeKey = everyKey + ":" + topicName;
>                                     if (everyKey.contains(findKey)) {
>                                         try {
>                                             if (logExpressionMap.containsKey(completeKey)) {
>     
>                                                 Boolean eventFlag = (Boolean) logExpressionMap.get(completeKey).execute(logCompareMap);
>     
>                                                 if (eventFlag) {  // 成功，放入 比较 compareMap true，放入缓存 MapState 状态为ture
>                                                     if (!expressionState.contains(completeKey)) {
>                                                         String expressionStateValue = expressionState.get(completeKey);
>     
>                                                         // 直接插入或者覆盖为 true
>                                                         expressionState.put(completeKey, "true");
>     
>                                                         // 对应日志匹配的结果 （A:XXXXX）
>                                                         String logType = completeKey.split(":")[2];
>     
>                                                         // 查询另一个表达式状态
>                                                         typeExpressionMap.put(logType, expressionStateValue);
>                                                     }
>     
>                                                 } else {
>                                                     // 匹配失败，放入compareMap false，放入缓存 MapState 状态为false
>                                                     expressionState.put(completeKey, "false");
>     
>                                                 }
>                                             }
>     
>                                         } catch (Exception ex) {
>     
>                                         }
>                                     } else {
>                                         String otherLogType = completeKey.split(":")[2];
>                                         // 其余的一个或者多个条件 去MapState里面去查询
>     
>                                         if (expressionState.contains(completeKey)) { //存在，然后拿到值
>                                             String othersCompleteKeyValue = expressionState.get(completeKey);
>     
>                                             typeExpressionMap.put(otherLogType, othersCompleteKeyValue);
>     
>                                         } else { // 不存在就是false
>     
>                                             typeExpressionMap.put(otherLogType, "false");
>                                         }
>     
>                                     }
>                                 }
>     
>     
>                                 // 在这里判断 typeExpressionMap的结果是否匹配
>                                 Boolean sequenceFlag = (Boolean) sequenceExpressionMap.get(topicName).execute(typeExpressionMap);
>                                 if (sequenceFlag) {
>                                     //todo 添加上匹配的topicName ，然后下发
>                                     logJson.put("topicName", topicName);
>                                     out.collect(logJson);
>                                 }
>                             }
>                         } else {
>                             System.out.println("预处理函数key对应valueList为null，findKey=" + findKey);
>                         }
>     
>                     } else {
>     
>                         System.out.println("预处理函数不包含这个key......" + findKey);
>                     }
>     
>                 }
>             }).uid("flatMap");
>     
>             //todo 6,将日志跟对应的场景表达式拼接，整合在一起
>     
>             DataStream<String> jsonArrayDstream = topicNameDstream.map(new MapFunction<JSONObject, JSONArray>() {
>     
>                 @Override
>                 public String map(JSONObject logValue) throws Exception {
>     
>                     //todo 遍历 日志，然后拼接在一起
>                     JSONArray eventsJsonArray = new JSONArray();
>     
>                     try {
>                         Map<String, TopicInfo> topicName2TopicInfoMap = RedisTableData.topicName2TopicInfoMap;
>     
>                         String topicName = logValue.getString("topicName");
>                         if (topicName2TopicInfoMap.size() > 0 && topicName2TopicInfoMap.containsKey(topicName)) {
>     
>                             // 拿到场景表达式
>                             TopicInfo topicInfoBean = topicName2TopicInfoMap.get(topicName);
>     
>                             // 遍历场景表达式
>                             List<String> interFaceKeyList = topicInfoBean.getInterFaceKeyList();
>     
>                             //TODO 拿 hbaseLogBean 取出需要的参数 ,返回一个数组
>     
>     
>                             String phoneTypeOfLog = logValue.getString("phoneType");
>                             String startInterfaceKeyStr = RedisTableData.startUpJsonOfProJson.get(phoneTypeOfLog).getAsString();
>                             Iterator<String> hbaseLogBeanIter = logValue.keySet().iterator();
>     
>     
>                             for (String interfaceKeyEntry : interFaceKeyList) {
>                                 JSONObject jsonTemp = new JSONObject();
>                                 //todo  获取场景表达式的 表达式接口 -- AAE , AAF
>                                 String thisInterfaceKey = interfaceKeyEntry.split(":")[1];
>     
>     
>                                 //todo 遍历hbaseBean
>                                 while (hbaseLogBeanIter.hasNext()) {
>     
>                                     String keyOfHbase = hbaseLogBeanIter.next();
>     
>                                     //todo 取出这个接口前缀跟 FFFFFF前缀 跟 启动接口前缀的数据
>                                     if (keyOfHbase.contains("_")) {
>                                         String[] prefixSplit = keyOfHbase.split("_");
>                                         String prefixKey = prefixSplit[0];
>                                         if (prefixKey.equals(thisInterfaceKey)
>                                                 || prefixKey.equals(Constaints.START_INTERFACE_PREFIX)
>                                                 || prefixKey.equals(startInterfaceKeyStr)) {
>     
>                                             String fieldKey = keyOfHbase.substring(keyOfHbase.indexOf("_") + 1);
>                                             String realValue = logValue.getString(keyOfHbase);
>                                             if (!jsonTemp.containsKey(fieldKey)) {
>                                                 jsonTemp.put(fieldKey, realValue);
>                                             }
>                                         }
>     
>                                     }
>                                 }
>                                 eventsJsonArray.add(jsonTemp);
>                             }
>     
>                         }
>     
>                     } catch (Exception e) {
>     
>                         e.printStackTrace();
>                     }
>     
>     
>                     return eventsJsonArray.toJSONString();
>                 }
>     
>             });
>             ParameterTool parameterTool = ParameterTool.fromArgs(args);
>             if (parameterTool.getNumberOfParameters() < 5) {
>                 System.out.println("Missing parameters!" +
>                         "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
>                         "--bootstrap.servers <kafka brokers> " +
>                         "--zookeeper.connect <zk quorum> --group.id <some id>");
>                 return;
>             }Properties pros = parameterTool.getProperties();
>     
>             //todo 7,sink输出到kafka
>             jsonArrayDstream.addSink(new FlinkKafkaProducer010<String>(
>                     "dianyou_wx_test",
>                     new SimpleStringSchema(),
>                     pros));
>             dataStream.print();
>     
>             env.execute("startExecute");
>         }
>     
>     
>     }