title: KSQL Group By to drop previous values and only use the LAST
body: <p>I have a Kafka topic "events" which records user image votes and has json in the following structure:</p>
  
  <pre><code>{"category":"image","action":"vote","label":"amsterdam","ip":"1.1.1.1","value":2}
  </code></pre>
  
  <p>I need to receive on another topic the sum of all votes for the label (e.g. amsterdam) but drop any votes that came from the same IP address using only the last vote. This topic should have json in this format:</p>
  
  <pre><code>{label:”amsterdam”,SCORE:8,TOTAL:3}
  </code></pre>
  
  <p>SCORE is a sum of all votes and TOTAL is the number of votes counted. </p>
  
  <p>The solution I made creates a stream from the topic events:</p>
  
  <pre><code>CREATE STREAM st_events
    (CATEGORY STRING, ACTION STRING, LABEL STRING, VALUE BIGINT, IP STRING)
    WITH (KAFKA_TOPIC='events', VALUE_FORMAT='JSON');
  </code></pre>
  
  <p>Then, I create a table  tb_votes which calculates the score and total for each label and IP address:</p>
  
  <pre><code>CREATE TABLE tb_votes WITH (KAFKA_TOPIC='tb_votes', PARTITIONS=1, REPLICAS=1) AS SELECT
    st_events.LABEL "label", SUM(st_events.VALUE-1) "score", CAST(COUNT(*) AS BIGINT) "total"
  FROM st_events
  WHERE
      st_events.category='image' AND st_events.action='vote'
  GROUP BY st_events.label, st_events.ip
  EMIT CHANGES;
  </code></pre>
  
  <p>The problem is that instead of dropping all the previous votes coming from the same ip address for the same image, Kafka uses all of them. This makes sense as it is a GROUP BY.</p>
  
  <p>Any idea how to "drop" all previous votes and only use the latest values for an image/ IP?</p>