Flink cep监控 Flink任务日志

- 日理万妓 2022-10-08 11:11 192阅读 0赞

# 1, 配置flink的log4j（idea本地测试） #

#log4j.rootLogger=info,console
    #log4j.appender.console=org.apache.log4j.ConsoleAppender
    #log4j.appender.console.layout=org.apache.log4j.PatternLayout
    #log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    
    
    
    log4j.rootLogger=WARN,kafka
    # for package com.demo.kafka, log would be sent to kafka appender.
    log4j.logger.kafka=WARN
    log4j.logger.org.apache.kafka=WARN
    log4j.logger.org.apache.zookeeper=WARN
    log4j.logger.org.I0Itec.zkclient=WARN
    ## appender console
    log4j.appender.console=org.apache.log4j.ConsoleAppender
    log4j.appender.console.target=System.out
    log4j.appender.console.layout=org.apache.log4j.PatternLayout
    log4j.appender.console.layout.ConversionPattern=%d (%t) [%p - %l] %m%n
    
    ## appender kafka
    log4j.appender.kafka=org.apache.kafka.log4jappender.KafkaLog4jAppender
    
    #指定topic
    log4j.appender.kafka.topic=aaa
    
    
    log4j.appender.kafka.brokerList=192.168.6.34:9092,192.168.6.35:9092,192.168.6.36:9092
    
    
    log4j.appender.kafka.compressionType=none
    log4j.appender.kafka.syncSend=false
    log4j.appender.kafka.layout=org.apache.log4j.PatternLayout
    log4j.appender.kafka.ThresholdFilter.level=INFO
    log4j.appender.kafka.ThresholdFilter.onMatch=ACCEPT
    log4j.appender.kafka.ThresholdFilter.onMismatch=DENY
    
    #log4j.appender.kafka.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% - %m%n
    log4j.appender.kafka.layout.ConversionPattern=%p||%c{1}||%L||%d{yyyy-MM-dd HH:mm:ss}||%m%n

# 2, 集群配置log4j.properties #

rootLogger.appenderRef.rolling.ref = RollingFile
    rootLogger.appenderRef.kafka.ref = Kafka
    rootLogger.level = WARN
    
    
    # The following lines keep the log level of common libraries/connectors on
    # log level INFO. The root logger does not override this. You have to manually
    # change the log levels here.
    logger.akka.name = akka
    logger.akka.level = INFO
    logger.kafka.name= org.apache.kafka
    logger.kafka.level = INFO
    logger.hadoop.name = org.apache.hadoop
    logger.hadoop.level = INFO
    logger.zookeeper.name = org.apache.zookeeper
    logger.zookeeper.level = INFO
    logger.pupu.name = com.pupu
    logger.pupu.level = DEBUG
    
    appender.rolling.type = RollingFile
    appender.rolling.name = RollingFile
    appender.rolling.append = false
    appender.rolling.fileName = ${sys:log.file}
    appender.rolling.filePattern = ${sys:log.file}.%i
    appender.rolling.layout.type = PatternLayout
    appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    appender.rolling.policies.type = Policies
    appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
    appender.rolling.policies.size.size = 10MB
    appender.rolling.strategy.type = DefaultRolloverStrategy
    appender.rolling.strategy.max = 10
    
    
    # Suppress the irrelevant (wrong) warnings from the Netty channel handler
    logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
    logger.netty.level = OFF
    
    # kafka appender config
    appender.kafka.type=Kafka
    appender.kafka.name=Kafka
    appender.kafka.topic=aaa
    appender.kafka.property.type=Property
    appender.kafka.property.name=bootstrap.servers
    appender.kafka.property.value=192.168.6.34:9092,192.168.6.35:9092,192.168.6.36:9092
    ## kafka的输出的日志pattern
    appender.kafka.layout.type=PatternLayout
    appender.kafka.layout.pattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% - %m%n

解释：这里是设置了每个文件大小为10M，保留10个日志文件，然后日志增量输出到 topic为‘aaa’里面去

数据格式为：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

# 3, 为了方面分析，我这里设置数据以'||'分割 #

日志为：

> ERROR||LogTest||53||2021-06-21 17:22:22||在算子里面打印日志:\{"canal\_type":"INSERT","hgjgjfgdf":"$$$","jhgjhgj,k":"555","order\_number":11,"gfhgfhghfg":"5%$%","gdfgfd":"3344","jghhfdgfdgdg":"INSERT"\},11

# 4，开发cep分析kafka数据代码     #

解释：

这里的代码很简单，遇到ERROR的错误日志就打印报警，这里只做一个演示而已

public class CepListenFlinkLogsDemo2 {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
    
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            long timestampParam = parameterTool.getLong("timestamp", 0L);
            String checkPointParam = parameterTool.get("checkpoint", "");
            String patternParam = parameterTool.get("prod", "");
    //
    
    
            Properties consumerProperties = getConsumerProperties();
            FlinkKafkaConsumer<JSONObject> kafkaConsumer = new FlinkKafkaConsumer<JSONObject>(
                    "aaa",
                    new StringDeserializationSchema(),
                    consumerProperties);
    
            //todo 策略模式
            kafkaConsumer.setStartFromEarliest();
            DataStreamSource<JSONObject> sourceStream = env.addSource(kafkaConsumer);
            sourceStream.print("==>");
    
            Pattern<JSONObject, ?> pattern = Pattern
                    .<JSONObject>begin("start").where(new RichIterativeCondition<JSONObject>() {
                        @Override
                        public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                            String log_type = value.getString("log_type");
                            if ("ERROR".equals(log_type)) {
                                return true;
                            }
                            return false;
                        }
                    });
    
            PatternStream patternStream = CEP.pattern(sourceStream.keyBy(new KeySelector<JSONObject, String>() {
                @Override
                public String getKey(JSONObject value) throws Exception {
                    return value.getString("task_name");
                }
            }), pattern);
    
    
            patternStream.select(new RichPatternSelectFunction<JSONObject, String>() {
                @Override
                public String select(Map<String, List<JSONObject>> pattern) throws Exception {
                    String start =  pattern.containsKey("start") ? pattern.get("start").get(0).toJSONString() : "" ;
                    return start;
                }
            }).print("---------->");
    
    
    //        sourceStream.print("==>");
    
    //        Pattern<JSONObject, JSONObject> within = Pattern.<JSONObject>begin("begin", AfterMatchSkipStrategy.skipToFirst("begin"))
    //                .where(new CommonFunction("operation_type=='1'&&ip =='武汉'"))
    //                .within(Time.days(3));
    //
    
    
            try {
    
                /*PatternStream patternStream = CEP.pattern(sourceStream, within);
    
                //todo 这里要不要做动态生成
    
                SingleOutputStreamOperator select = patternStream.flatSelect(new PatternFlatSelectFunction<JSONObject, List<JSONObject>>() {
    
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<List<JSONObject>> out) throws Exception {
                        if (selectArray.size() > 1) {
    
                            List<JSONObject> list = new ArrayList<>();
                            for (int i = 0; i < selectArray.size(); i++) {
                                JSONObject select = selectArray.getJSONObject(i);
                                String conditionName = select.getString("condition_name");
                                String getListOrget = select.getString("do");
                                List<JSONObject> begin = map.get(conditionName);
                                list.addAll(begin);
    
                            }
                            out.collect(list);
                        } else {
                            JSONObject select = selectArray.getJSONObject(0);
                            String conditionName = select.getString("condition_name");
                            List<JSONObject> rs = map.get(conditionName);
                            out.collect(rs);
                        }
    
    
                    }
    
                });*/
    //            select.print("out输出:");
    //            select.addSink(new HBaseWriter());
    
                //todo 侧流输出
                /*SingleOutputStreamOperator result = pattern1.process(
                        new PatternProcessFunction<LoginEvent, String>() {
                            @Override
                            public void processMatch(Map<String, List<LoginEvent>> map, Context context, Collector<String> out) throws Exception {
                                List<LoginEvent> begin1 = map.get("begin");
    
                                String begin = map.get("begin").get(0).toString();
                                String next = map.get("next").get(0).toString();
                                out.collect("begin=" + begin + "_____" + "next=" + next);
                            }
                        });*/
            } catch (Exception e) {
                e.printStackTrace();
            }
    
            try {
                env.execute("aaa");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public static Properties getConsumerProperties() {
            Properties properties = new Properties();
            properties.put("bootstrap.servers", Constants.BOOTSTRAP_SERVERS);
            properties.put("group.id", "aaa");
            properties.setProperty("flink.partition-discovery.interval-millis", 300 * 1000 + "");  // 自动发现消费的partition变化
            return properties;
        }
    
        public static Properties getSinkProperties() {
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", Constants.BOOTSTRAP_SERVERS);
            props.setProperty("buffer.memory", Constants.BUFFER_MEMORY);
            props.setProperty("batch.size", Constants.BATCH_SIZE);
            props.setProperty("linger.ms", Constants.LINGER_MS);
            props.setProperty("max.request.size", Constants.MAX_REQUEST_SIZE);
            props.setProperty("acks", Constants.ACKS);
            props.setProperty("retries", Constants.RETRIES);
            props.setProperty("retry.backoff.ms", Constants.RETRY_BACKOFF_MS);
            return props;
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221005/ed366ed0ec4e40db8be46a471570b6e9.png