flink1.12.x Transformation-map、flatMap算子使用lambda表达式，返回tuple发生泛型擦除

作者：马育民 • • 阅读：10071

提出问题

package map;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
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;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class TestMap泛型擦除 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        DataStreamSource<String> dss = env.addSource(fkc);

        // 执行时报错
        SingleOutputStreamOperator<Tuple2<String,Integer>> soso = dss.map(s-> Tuple2.of(s,1));

        soso.print();
        env.execute();
    }


}

执行时报错，如下：

The return type of function 'main(TestMap���Ͳ���.java:29)' could not be determined automatically, due to type erasure. You can give type information hints by using the returns(...) method on the result of the transformation call, or by letting your function implement the 'ResultTypeQueryable' interface.

分析问题

map()、flatMap() 这样的方法，方法声明有 泛型，被 Java 编译器编译后，泛型信息被擦除了。

使用 lambda表达式，返回 Tuple 元祖时，无法自动推断输出的类型

解决

方式一：使用lambda表达式

使用lambda表达式，需要显式地指定类型信息，如下：

SingleOutputStreamOperator<Tuple2<String,Integer>> soso = dss.map(s-> Tuple2.of(s,1))
        // 指定返回类型
        .returns(Types.TUPLE(Types.STRING,Types.INT));

实现类：

static class MyMap implements MapFunction<String, Tuple2<String, Integer>> {
    @Override
    public Tuple2<String, Integer> map(String s) throws Exception {
        return Tuple2.of(s,1);
    }
}

方式二：使用实现类

SingleOutputStreamOperator<Tuple2<String,Integer>> soso = dss.map(new MyMap());

方式三：使用匿名类

SingleOutputStreamOperator<Tuple2<String,Integer>> soso = dss.map(new MapFunction<String, Tuple2<String, Integer>>() {
        @Override
        public Tuple2<String, Integer> map(String s) throws Exception {
            return Tuple2.of(s,1);
        }
    });

参考

https://blog.csdn.net/m0_65025800/article/details/123662009

原文出处：/show_1IX43otYmN7K.html

吉ICP备15003043号