如何实现数据集批处理的第一个元组字段元素的keyBy

Question

我正在尝试将我的应用程序从flink流处理转换为flink批处理。

对于flink数据流，我从包含多个JSON对象的预定义文件中读取字符串，并执行从JSON对象到tuple3收集器的flatmap（第一个元素--来自JSON对象的一个字段，第二个元素--来自JSON对象的另一个字段，第三个元素--实际的JSON对象数据）。

DataStream<Tuple3<String, Integer, ObjectNode>> transformedSource = source.flatMap(new FlatMapFunction<String, Tuple3<String, Integer, ObjectNode>>() {
                @Override
                public void flatMap(String value, Collector<Tuple3<String, Integer, ObjectNode>> out) throws Exception {
                    ObjectNode record = mapper.readValue(value, ObjectNode.class);
                    JsonNode customer = record.get("customer");
                    JsonNode deviceId = record.get("id");
                                       if (customer != null && deviceId != null) {
                        out.collect(Tuple3.of(customer.asText(), deviceId.asInt(), record));
                    }
                }
            });

然后，在窗口中执行元组的first和element的keyBy。

WindowedStream<Tuple3<String, Integer,ObjectNode>, Tuple, TimeWindow> combinedData = transformedSource
            .keyBy(0, 1)
            .timeWindow(Time.seconds(5));

对于flink批处理，如何做数据集批处理的KeyBy，数据集中是否有KeyBy的等价方法

DataSet<String> source = env.readTextFile("file:///path /to/ file");


DataSet<Tuple3<String, Integer, ObjectNode>> transformedSource = source.flatMap(new FlatMapFunction<String, Tuple3<String, Integer, ObjectNode>>() {
                @Override
                public void flatMap(String value, Collector<Tuple3<String, Integer, ObjectNode>> out) throws Exception {
                    ObjectNode record = mapper.readValue(value, ObjectNode.class);
                    JsonNode customer = record.get("customer");
                    JsonNode deviceId = record.get("id");
                                       if (customer != null && deviceId != null) {
                        out.collect(Tuple3.of(customer.asText(), deviceId.asInt(), record));
                    }
                }
            });

Answer 1

groupBy似乎是您正在寻找的方法