Flink 映射流 csv 文件到 Tuple

Question

我正在尝试将已由 Flink 使用并由 Kafka 生成的 CSV 文件映射到 Tuple4。我的 CSV 文件有 4 列，我想将每一行映射到一个 Tuple4。问题是我不知道如何实现 map() 和 csv2Tuple 函数。

这是我卡住的地方：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

ParameterTool parameterTool = ParameterTool.fromArgs(ARGS);

DataStreamSource<String> myConsumer = env.addSource(new FlinkKafkaConsumer082<>(parameterTool.getRequired("topic"),
            new SimpleStringSchema(), parameterTool.getProperties()));

DataStream<Tuple4<Integer, Integer, Integer, Integer>> streamTuple = myConsumer.map(new csv2Tuple());
public static class csv2Tuple implements MapFunction<...> {public void map(){...}}

我还想将元组中的项目从 String 解析为 Integer。

Answer 1

假设您将每一行 csv 文件作为 Kafka 消息生成并使用 Flink Kafka 连接器使用它，您只需使用 , 拆分每条使用的消息（因为它是一个 csv 文件）。

DataStream<Tuple4<Integer, Integer, Integer, Integer,>> streamTuple = myConsumer.map(new MapFunction<String, Tuple4<Integer, Integer, Integer, Integer>>() {
            @Override
            public Tuple4<Integer, Integer, Integer, Integer> map(String str) throws Exception {
                String[] temp = str.split(",");
                return new Tuple4<>(
                        Integer.parseInt(temp[0]),
                        Integer.parseInt(temp[1]),
                        Integer.parseInt(temp[2]),
                        Integer.parseInt(temp[3])
                );

            }
        });