Flink在《Kafka》中克服坏消息：“毒药消息”

Question

我第一次试着让它工作，所以请容忍我。我正在尝试学习Kafka的检查点设置和处理“错误”消息，在不丢失状态的情况下重新启动。

用例：使用检查点。从Kafka那里读取一个整数流，保持一个连续的和。如果读到“坏”Kafka消息，请重新启动应用程序，跳过“坏”消息，保持状态。我的流看起来像这样：

set1,5
set1,7
set1，foobar
set1,6

我希望我的应用程序保留它看到的整数的运行总和，如果它崩溃而不丢失状态，则重新启动，因此应用程序行为/运行总和将是：
5、
12、
应用程序崩溃并重新启动、读取检查点
18等。

但是，我发现当我的应用程序重新启动时，它一直在读取错误的“fobar”消息并且无法通过它。下面的源代码。当我尝试将“fobar”解析为整数时，映射器会爆炸。如何修改应用程序以绕过“毒药”消息？

    env.enableCheckpointing(1000L);   


   env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); 
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); 
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500L); 
        env.getCheckpointConfig().setCheckpointTimeout(10000); 
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); 
        env.setStateBackend(new 
        FsStateBackend("hdfs://mymachine:9000/flink/checkpoints")); 

        Properties properties = new Properties(); 
        properties.setProperty("bootstrap.servers", BROKERS); 
        properties.setProperty("zookeeper.connect", ZOOKEEPER_HOST); 
        properties.setProperty("group.id", "consumerGroup1"); 

        FlinkKafkaConsumer08 kafkaConsumer = new FlinkKafkaConsumer08<>(topicName, 
        new SimpleStringSchema(), properties); 
        DataStream<String> messageStream = env.addSource(kafkaConsumer); 

        DataStream<Tuple2&lt;String,Integer>> sums = messageStream 
          .map(new NumberMapper()) 
          .keyBy(0) 
          .sum(1);  
          sums.print(); 


                private static class NumberMapper implements 
        MapFunction<String,Tuple2<String,Integer>> { 
                        public Tuple2<String,Integer> map(String input) throws Exception { 
                                return parseData(input); 
                        } 

                        private Tuple2<String,Integer> parseData(String record) { 

                                String[] tokens = record.toLowerCase().split(","); 

                                // Get Key 
                                String key = tokens[0]; 

                                // Get Integer Value 
                                String integerValue = tokens[1]; 
                                System.out.println("Trying to Parse=" + integerValue); 
                                Integer value = Integer.parseInt(integerValue); 

                                // Build Tuple
                                return new Tuple2<String,Integer>(key, value); 
                        } 

                } 

Answer 1

您可以将NumberMapper更改为FlatMap并过滤掉无效元素：

java prettyprint-override">private static class NumberMapper implements FlatMapFunction<String, Tuple2<String, Integer>> { 
        public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) throws Exception { 
                Optional<Tuple2<String, Integer>> optionalResult = parseData(input); 

                optionalResult.ifPresent(collector::collect);
        } 

        private Optional<Tuple2<String, Integer>> parseData(String record) { 

                String[] tokens = record.toLowerCase().split(","); 

                // Get Key 
                String key = tokens[0]; 

                // Get Integer Value 
                String integerValue = tokens[1]; 

                try {
                        Integer value = Integer.parseInt(integerValue); 
                        // Build Tuple
                        return Optional.of(Tuple2.of(key, value)); 
                } catch (NumberFormatException e) {
                        return Optional.empty();
                }
        } 
}