title: How to calculate two sum on the same dataset with Apache Flink
body: <p>I have a simple stream if data which is of this form:</p>
  
  <pre><code>id | name | eventType | eventTime
  ----------------------------------
  1    A       PLAY        (ts of when the client fired the event)
  1    B       IMPRESSION
  2    A       CLICK
  </code></pre>
  
  <p>The end goal is to calculate the sum of event of the eventType CLICK divided by the sum of eventType of the type IMPRESSION grouped by ID and NAME for a tumbling window of 60 seconds.</p>
  
  <p>in pure SQL it would look like</p>
  
  <pre><code>SELECT d.id, d.name, d.impressionCount, d.clickCount,  d.clickCount / d.impressionCount * 100.0 FROM
  ( SELECT i.id, i.name, count(*) as clickCount, c.impressionCount from events as i
  LEFT JOIN
   (
      SELECT id, name, count(*) as impressionCount from events WHERE event_type = 'IMPRESSION' GROUP BY id,name
   ) as c
  ON i.id = c.id and i.name = c.name
  WHERE event_type = 'CLICK' 
   GROUP BY i.id, i.name
  ) as d
  </code></pre>
  
  <p>So I first need to create a column with the number clicks and a new column with the number of impression and then i use that table to do a division.</p>
  
  <p>My question is.. what is the best to do this with Flink APis ? I have attempted to do this:</p>
  
  <pre><code>Table clickCountTable = eventsTable
                  .where("eventType = 'CLICK'")
                  .window(Tumble.over("1.minute").on("eventTime").as("minuteWindow"))
                  .groupBy("id, name, minuteWindow")
                  .select("concat(concat(id,'_'), name) as id, eventType.count as clickCount, minuteWindow.rowtime as minute");
  </code></pre>
  
  <p>and same for the impression and then I join this two table. But I do not get the right result and I'm not sure this is the best way to achieve what I want to do using tubling window.</p>
  
  <p><strong>EDIT</strong>:</p>
  
  <p>This is how I transform the stream into tables:</p>
  
  <pre><code>env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  
  [.....]
  DataStream&lt;EventWithCount&gt; eventStreamWithTime = eventStream
              .assignTimestampsAndWatermarks(new AscendingTimestampExtractor&lt;EventWithCount&gt;() {
                  @Override
                  public long extractAscendingTimestamp(EventWithCount element) {
                      try {
                          DateFormat df1 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSSSS");
                          Date parsedDate = df1.parse(element.eventTime);
                          Timestamp timestamp = new java.sql.Timestamp(parsedDate.getTime());
                          return timestamp.getTime();
                      } catch (Exception e) {
                          throw new RuntimeException(e.getMessage());
                      }
                  }});
  tEnv.fromDataStream(eventStreamWithTime, "id, name, eventType, eventTime.rowtime");
  tEnv.registerTable("Events", eventsTable);
  </code></pre>