title: apache flink 0.10 how to get the first occurence of a composite key from an unbounded input dataStream?
body: <p>I am a newbie with apache flink. I have an unbound data stream in my input (fed into flink 0.10 via kakfa).</p>
  <p>I want to get the 1st occurence of each primary key (the primary key is the contract_num and the event_dt). <br/>
  These &quot;duplicates&quot; occur nearly immediately after each other.
  The source system cannot filter this for me, so flink has to do it.</p>
  <p>Here is my input data:</p>
  <pre class="lang-none prettyprint-override"><code>contract_num, event_dt, attr 
  A1, 2016-02-24 10:25:08, X
  A1, 2016-02-24 10:25:08, Y
  A1, 2016-02-24 10:25:09, Z
  A2, 2016-02-24 10:25:10, C
  </code></pre>
  <p>Here is the output data I want:</p>
  <pre class="lang-none prettyprint-override"><code>A1, 2016-02-24 10:25:08, X
  A1, 2016-02-24 10:25:09, Z
  A2, 2016-02-24 10:25:10, C
  </code></pre>
  <p>Note the 2nd row has been removed as the key combination of A001 and '2016-02-24 10:25:08' already occurred in the 1st row.</p>
  <p>How can I do this with flink 0.10?</p>
  <p>I was thinking about using <code>keyBy(0,1)</code> but after that I don't know what to do!</p>
  <p>(I used joda-time and org.flinkspector to setup these tests).</p>
  <pre class="lang-java prettyprint-override"><code>@Test
  public void test() {
      DateTime threeSecondsAgo = (new DateTime()).minusSeconds(3);
      DateTime twoSecondsAgo = (new DateTime()).minusSeconds(2);
      DateTime oneSecondsAgo = (new DateTime()).minusSeconds(2);
  
      DataStream&lt;Tuple3&lt;String, Date, String&gt;&gt; testStream =
              createTimedTestStreamWith(
                      Tuple3.of(&quot;A1&quot;, threeSecondsAgo.toDate(), &quot;X&quot;))
              .emit(Tuple3.of(&quot;A1&quot;, threeSecondsAgo.toDate(), &quot;Y&quot;), after(0, TimeUnit.NANOSECONDS))
              .emit(Tuple3.of(&quot;A1&quot;, twoSecondsAgo.toDate(), &quot;Z&quot;), after(0, TimeUnit.NANOSECONDS))
              .emit(Tuple3.of(&quot;A2&quot;, oneSecondsAgo.toDate(), &quot;C&quot;), after(0, TimeUnit.NANOSECONDS))
              .close();
      
      testStream.keyBy(0,1);
  }
  </code></pre>