Flink keyBy groping issue

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番目に出力します（このキー処理の合計が2になります）。その空きを別のキープロセスに送り、1を出力します。これは私にとって適切な出力のようです。以下のコメントパー

EDIT

、ここであなたに所望の出力を与えるコードは次のとおりです。

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(); 
}