Flink keyBy 分组问题

Question

我对 Flink 还是很陌生。我有这段代码可以对输入的 JSON 进行映射、分组和求和。

这与字数统计示例非常相似。

我希望得到(vacant,1) (occupied,2)

但是，由于某种原因，我收到了(occupied,1) (vacant,1) (occupied,2)

  public static void main(String[] args) throws Exception {
        String s = "{\n" +
                "    \"Port_128\": \"occupied\",\n" +
                "    \"Port_129\": \"occupied\",\n" +
                "    \"Port_120\": \"vacant\"\n" +
                "\n" +
                "}";
        StreamExecutionEnvironment env = 
        StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> in = env.fromElements(s);
        SingleOutputStreamOperator<Tuple2<String, Integer>> t = 
        in.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> 
            collector) throws Exception {
                ObjectMapper mapper = new ObjectMapper();
                JsonNode node = mapper.readTree(s);
                node.elements().forEachRemaining(v -> {
                    collector.collect(new Tuple2<>(v.textValue(), 1));
                });

            }
        }).keyBy(0).sum(1);

        t.print();
        env.execute();

Answer 1

运行你的代码，我得到：

10/19/2017 11:27:38 Keyed Aggregation -> Sink: Unnamed(1/1) switched to RUNNING 
(occupied,1)
(occupied,2)
(vacant,1)
10/19/2017 11:28:03 Keyed Aggregation -> Sink: Unnamed(1/1) switched to FINISHED 

这与您的输出略有不同，但很重要。原因是代码在接收数据时输出每个键的总和，所以首先它被第一个占用（输出 1），然后是第二个（输出这个键控进程的总和现在是 2），然后将空置发送到另一个键控进程并输出 1。所以这对我来说似乎是正确的输出。

编辑

根据下面的评论，这里是可以为您提供所需输出的代码：

public static void main(String[] args) throws Exception {
  String s = "{\n" +
      "    \"Port_128\": \"occupied\",\n" +
      "    \"Port_129\": \"occupied\",\n" +
      "    \"Port_120\": \"vacant\"\n" +
      "\n" +
      "}";
  ExecutionEnvironment env =
      ExecutionEnvironment.getExecutionEnvironment();
  DataSet<String> in = env.fromElements(s);
  AggregateOperator<Tuple2<String, Integer>> t =
      in.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>>
            collector) throws Exception {
          ObjectMapper mapper = new ObjectMapper();
          JsonNode node = mapper.readTree(s);
          node.elements().forEachRemaining(v -> {
            collector.collect(new Tuple2<>(v.textValue(), 1));
          });

        }
      }).groupBy(0).sum(1);

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