title: How can I improve my count distinct for data stream implementation in Flink?
body: <p>I have this program in Flink (Java) which count the distinct words in a data stream. I implemented using the example of count words and them I applied another window with the same time to evaluate the distinct values. The program is working fine. However, I am concerned that I am using two windows to process a distinct count. The first window counts the number of words and the second window I switch the number of words to be <code>1</code> and the word to be the second element of a <code>Tuple2</code>. Them I count the number of keys. This is the input and out put of my program:</p>
  
  <pre><code>// input:
  aaa
  aaa
  bbb
  ccc
  bbb
  aaa
  output:
  (3,bbb-ccc-aaa)
  </code></pre>
  
  <p>If I remove the second window it shows all the evaluating for each key and saves the state of the previous window.</p>
  
  <pre><code>// input:
  aaa
  aaa
  bbb
  ccc
  bbb
  aaa
  // output:
  3&gt; (1,bbb)
  3&gt; (2,bbb-aaa)
  3&gt; (3,bbb-aaa-ccc)
  // wait the first window to be evaluated.
  // input:
  aaa
  aaa
  bbb
  ccc
  bbb
  aaa
  // output:
  3&gt; (4,bbb-aaa-ccc-ccc)
  3&gt; (5,bbb-aaa-ccc-ccc-bbb)
  3&gt; (6,bbb-aaa-ccc-ccc-bbb-aaa)
  </code></pre>
  
  <p>My program:</p>
  
  <pre><code>public class WordCountDistinctSocketFilterQEP {
      public WordCountDistinctSocketFilterQEP() throws Exception {
  
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
          // @formatter:off
          env.socketTextStream("localhost", 9000)
                  .flatMap(new SplitterFlatMap())
                  .keyBy(new MyKeySelector())
                  .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                  .reduce(new CountReduceFunction())
                  .map(new SwapMapFunction())
                  .keyBy(0)
                  .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // TESTING REMOVING THIS WINDOW
                  .reduce(new CountDistinctFunction())
                  .print();
          // @formatter:on
  
          String executionPlan = env.getExecutionPlan();
          System.out.println("ExecutionPlan ........................ ");
          System.out.println(executionPlan);
          System.out.println("........................ ");
          // dataStream.print();
  
          env.execute("WordCountDistinctSocketFilterQEP");
      }
  
      public static class SwapMapFunction implements MapFunction&lt;Tuple2&lt;String, Integer&gt;, Tuple2&lt;Integer, String&gt;&gt; {
          private static final long serialVersionUID = 5148172163266330182L;
  
          @Override
          public Tuple2&lt;Integer, String&gt; map(Tuple2&lt;String, Integer&gt; value) throws Exception {
              return Tuple2.of(1, value.f0);
          }
      }
  
      public static class SplitterFlatMap implements FlatMapFunction&lt;String, Tuple2&lt;String, Integer&gt;&gt; {
          private static final long serialVersionUID = 3121588720675797629L;
  
          @Override
          public void flatMap(String sentence, Collector&lt;Tuple2&lt;String, Integer&gt;&gt; out) throws Exception {
              for (String word : sentence.split(" ")) {
                  out.collect(new Tuple2&lt;String, Integer&gt;(word, 1));
              }
          }
      }
  
      public static class MyKeySelector implements KeySelector&lt;Tuple2&lt;String, Integer&gt;, String&gt; {
          private static final long serialVersionUID = 2787589690596587044L;
  
          @Override
          public String getKey(Tuple2&lt;String, Integer&gt; value) throws Exception {
              return value.f0;
          }
      }
  
      public static class CountReduceFunction implements ReduceFunction&lt;Tuple2&lt;String, Integer&gt;&gt; {
          private static final long serialVersionUID = 8541031982462158730L;
  
          @Override
          public Tuple2&lt;String, Integer&gt; reduce(Tuple2&lt;String, Integer&gt; value1, Tuple2&lt;String, Integer&gt; value2)
                  throws Exception {
              return Tuple2.of(value1.f0, value1.f1 + value2.f1);
          }
      }
  
      public static class CountDistinctFunction implements ReduceFunction&lt;Tuple2&lt;Integer, String&gt;&gt; {
          private static final long serialVersionUID = -7077952757215699563L;
  
          @Override
          public Tuple2&lt;Integer, String&gt; reduce(Tuple2&lt;Integer, String&gt; value1, Tuple2&lt;Integer, String&gt; value2)
                  throws Exception {
              return Tuple2.of(value1.f0 + value2.f0, value1.f1 + "-" + value2.f1);
          }
      }
  }
  </code></pre>