1

For data that is beeing streamed from out ticket-system we try to achieve the following

Get the number of open tickets grouped by status and customer. The simplified schema is as follows


 Field               | Type                      
-------------------------------------------------
 ROWTIME             | BIGINT           (system) 
 ROWKEY              | VARCHAR(STRING)  (system) 
 ID                  | BIGINT                    
 TICKET_ID           | BIGINT                    
 STATUS              | VARCHAR(STRING)           
 TICKETCATEGORY_ID   | BIGINT                    
 SUBJECT             | VARCHAR(STRING)           
 PRIORITY            | VARCHAR(STRING)           
 STARTTIME           | BIGINT                    
 ENDTIME             | BIGINT                    
 CHANGETIME          | BIGINT                    
 REMINDTIME          | BIGINT                    
 DEADLINE            | INTEGER                   
 CONTACT_ID          | BIGINT           

We want to use that data to get the number of tickets with a certain status (open, waiting,in progress, etc.) per customer. This data has to one message in another topic- The scheme could look like that

 Field               | Type                      
-------------------------------------------------
 ROWTIME             | BIGINT           (system) 
 ROWKEY              | VARCHAR(STRING)  (system) 
 CONTACT_ID          | BIGINT                    
 COUNT_OPEN          | BIGINT                    
 COUNT_WAITING       | BIGINT                    
 COUNT_CLOSED        | BIGINT                    

We plan to use this and other data to enrich the customer-information and publish the enriched dataset to an external system (eg elasticsearch)

It is pretty easy to get the first part - grouping the tickets by customer and status.

select contact_id,status count(*) cnt from tickets group by contact_id,status;

But now we are stuck - we get multiple rows/messages per customer, and we just don't know how to transform them into one message with the contact_id as the key.

We tried joins but all our tries led to nothing.

Example

Create table for all ticket with status "waiting" grouped by customers

create table waiting_tickets_by_cust with (partitions=12,value_format='AVRO')
as select contact_id, count(*) cnt from tickets where status='waiting' group by contact_id;

Rekey table for join

CREATE TABLE T_WAITING_REKEYED with WITH (KAFKA_TOPIC='WAITING_TICKETS_BY_CUST',
       VALUE_FORMAT='AVRO',
       KEY='contact_id');

Left (outer) joining that table with our customer table gets us all customers that have an tickets waiting.

select c.id,w.cnt wcnt from T_WAITING_REKEYED w left join CRM_CONTACTS c on w.contact_id=c.id;

But we would need all Customers, with the waiting count NULLED to use that result in another join with tickets in status PROCESSING. Since we only have customers with waiting, only gets us those that have values for both status.

ksql> select c.*,t.cnt from T_PROCESSING_REKEYED t left join cust_ticket_tmp1 c on t.contact_id=c.id;
null | null | null | null | 1
1555261086669 | 1472 | 1472 | 0 | 1
1555261086669 | 1472 | 1472 | 0 | 1
null | null | null | null | 1
1555064371937 | 1474 | 1474 | 1 | 1
null | null | null | null | 1
1555064371937 | 1474 | 1474 | 1 | 1
null | null | null | null | 1
null | null | null | null | 1
null | null | null | null | 1
1555064372018 | 3 | 3 | 5 | 6
1555064372018 | 3 | 3 | 5 | 6

So what is the correct approach to do this ?

This is KSQL 5.2.1

Thank you

Edit:

Here is some sample data

Created a TOPIC that limits the data to a test-account

CREATE STREAM tickets_filtered
  WITH (
        PARTITIONS=12,
        VALUE_FORMAT='JSON') AS
  SELECT id,
         contact_id,
subject,
status,

         TIMESTAMPTOSTRING(changetime, 'yyyy-MM-dd HH:mm:ss.SSS') AS timestring
  FROM tickets where contact_id=1472
  PARTITION BY contact_id;

00:06:44 1 $ kafkacat-dev -C -o beginning -t TICKETS_FILTERED
{"ID":2216,"CONTACT_ID":1472,"SUBJECT":"Test Bodenbach","STATUS":"closed","TIMESTRING":"2012-11-08 10:34:30.000"}
{"ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"waiting","TIMESTRING":"2019-04-16 23:07:01.000"}
{"ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"processing","TIMESTRING":"2019-04-16 23:52:08.000"}
Changing and adding something in the ticketing-system...
{"ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"waiting","TIMESTRING":"2019-04-17 00:10:38.000"}
{"ID":8952,"CONTACT_ID":1472,"SUBJECT":"another sync ticket","STATUS":"new","TIMESTRING":"2019-04-17 00:11:23.000"}
{"ID":8952,"CONTACT_ID":1472,"SUBJECT":"another sync ticket","STATUS":"close-request","TIMESTRING":"2019-04-17 00:12:04.000"}

We want to create a topic out of that data where the messages look like this

{"CONTACT_ID":1472,"TICKETS_CLOSED":1,"TICKET_WAITING":1,"TICKET_CLOSEREQUEST":1,"TICKET_PROCESSING":0}
2
  • Do you have some sample data you could share? Apr 15, 2019 at 12:26
  • Yes, i had to redact it a little bit but i added to the original request
    – iceman76
    Apr 16, 2019 at 22:17

1 Answer 1

2

(written up here too)

It's possible to do this by building a table (for state) and then an aggregate on that table.

  1. Set up the test data

    kafkacat -b localhost -t tickets -P <<EOF
    {"ID":2216,"CONTACT_ID":1472,"SUBJECT":"Test Bodenbach","STATUS":"closed","TIMESTRING":"2012-11-08 10:34:30.000"}
    {"ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"waiting","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"processing","TIMESTRING":"2019-04-16 23:52:08.000"}
    {"ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"waiting","TIMESTRING":"2019-04-17 00:10:38.000"}
    {"ID":8952,"CONTACT_ID":1472,"SUBJECT":"another sync ticket","STATUS":"new","TIMESTRING":"2019-04-17 00:11:23.000"}
    {"ID":8952,"CONTACT_ID":1472,"SUBJECT":"another sync ticket","STATUS":"close-request","TIMESTRING":"2019-04-17 00:12:04.000"}
    EOF
    
  2. Preview the topic data

    ksql> PRINT 'tickets' FROM BEGINNING;
    Format:JSON
    {"ROWTIME":1555511270573,"ROWKEY":"null","ID":2216,"CONTACT_ID":1472,"SUBJECT":"Test Bodenbach","STATUS":"closed","TIMESTRING":"2012-11-08 10:34:30.000"}
    {"ROWTIME":1555511270573,"ROWKEY":"null","ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"waiting","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ROWTIME":1555511270573,"ROWKEY":"null","ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"processing","TIMESTRING":"2019-04-16 23:52:08.000"}
    {"ROWTIME":1555511270573,"ROWKEY":"null","ID":8945,"CONTACT_ID":1472,"SUBJECT":"sync-test","STATUS":"waiting","TIMESTRING":"2019-04-17 00:10:38.000"}
    {"ROWTIME":1555511270573,"ROWKEY":"null","ID":8952,"CONTACT_ID":1472,"SUBJECT":"another sync ticket","STATUS":"new","TIMESTRING":"2019-04-17 00:11:23.000"}
    {"ROWTIME":1555511270573,"ROWKEY":"null","ID":8952,"CONTACT_ID":1472,"SUBJECT":"another sync ticket","STATUS":"close-request","TIMESTRING":"2019-04-17 00:12:04.000"}
    
  3. Register the stream

    CREATE STREAM TICKETS (ID INT, 
                          CONTACT_ID VARCHAR, 
                          SUBJECT VARCHAR, 
                          STATUS VARCHAR, 
                          TIMESTRING VARCHAR) 
            WITH (KAFKA_TOPIC='tickets', 
            VALUE_FORMAT='JSON');
    
  4. Query the data

    ksql> SET 'auto.offset.reset' = 'earliest';
    ksql> SELECT * FROM TICKETS;
    1555502643806 | null | 2216 | 1472 | Test Bodenbach | closed | 2012-11-08 10:34:30.000
    1555502643806 | null | 8945 | 1472 | sync-test | waiting | 2019-04-16 23:07:01.000
    1555502643806 | null | 8945 | 1472 | sync-test | processing | 2019-04-16 23:52:08.000
    1555502643806 | null | 8945 | 1472 | sync-test | waiting | 2019-04-17 00:10:38.000
    1555502643806 | null | 8952 | 1472 | another sync ticket | new | 2019-04-17 00:11:23.000
    1555502643806 | null | 8952 | 1472 | another sync ticket | close-request | 2019-04-17 00:12:04.000
    
  5. At this point we can use CASE to pivot the aggregates:

    SELECT CONTACT_ID, 
          SUM(CASE WHEN STATUS='new' THEN 1 ELSE 0 END) AS TICKETS_NEW, 
          SUM(CASE WHEN STATUS='processing' THEN 1 ELSE 0 END) AS TICKETS_PROCESSING, 
          SUM(CASE WHEN STATUS='waiting' THEN 1 ELSE 0 END) AS TICKETS_WAITING, 
          SUM(CASE WHEN STATUS='close-request' THEN 1 ELSE 0 END) AS TICKETS_CLOSEREQUEST ,
          SUM(CASE WHEN STATUS='closed' THEN 1 ELSE 0 END) AS TICKETS_CLOSED
      FROM TICKETS 
      GROUP BY CONTACT_ID;
    
      1472 | 1 | 1 | 2 | 1 | 1
    

    But, you'll notice that the answer isn't as expected. This is because we're counting all six input events.

    Let's look at a single ticket, ID 8945—this goes through three state changes (waiting -> processing -> waiting) which each get included in the aggregate. We can validate this as follows with a simple predicate:

    SELECT CONTACT_ID, 
          SUM(CASE WHEN STATUS='new' THEN 1 ELSE 0 END) AS TICKETS_NEW, 
          SUM(CASE WHEN STATUS='processing' THEN 1 ELSE 0 END) AS TICKETS_PROCESSING, 
          SUM(CASE WHEN STATUS='waiting' THEN 1 ELSE 0 END) AS TICKETS_WAITING, 
          SUM(CASE WHEN STATUS='close-request' THEN 1 ELSE 0 END) AS TICKETS_CLOSEREQUEST ,
          SUM(CASE WHEN STATUS='closed' THEN 1 ELSE 0 END) AS TICKETS_CLOSED
      FROM TICKETS 
      WHERE ID=8945
      GROUP BY CONTACT_ID;
    
    1472 | 0 | 1 | 2 | 0 | 0
    
  6. What we actually want is the current state for each ticket. So repartition the data on ticket ID:

    CREATE STREAM TICKETS_BY_ID AS SELECT * FROM TICKETS PARTITION BY ID;
    
    CREATE TABLE TICKETS_TABLE (ID INT, 
                          CONTACT_ID INT, 
                          SUBJECT VARCHAR, 
                          STATUS VARCHAR, 
                          TIMESTRING VARCHAR) 
            WITH (KAFKA_TOPIC='TICKETS_BY_ID', 
            VALUE_FORMAT='JSON',
            KEY='ID');
    
  7. Compare event stream vs current state

    • Event stream (KSQL Stream)

      ksql> SELECT ID, TIMESTRING, STATUS FROM TICKETS;
      2216 | 2012-11-08 10:34:30.000 | closed
      8945 | 2019-04-16 23:07:01.000 | waiting
      8945 | 2019-04-16 23:52:08.000 | processing
      8945 | 2019-04-17 00:10:38.000 | waiting
      8952 | 2019-04-17 00:11:23.000 | new
      8952 | 2019-04-17 00:12:04.000 | close-request
      
    • Current state (KSQL Table)

      ksql> SELECT ID, TIMESTRING, STATUS FROM TICKETS_TABLE;
      2216 | 2012-11-08 10:34:30.000 | closed
      8945 | 2019-04-17 00:10:38.000 | waiting
      8952 | 2019-04-17 00:12:04.000 | close-request
      
  8. We want an aggregate of the table—we want to run the same SUM(CASE…)…GROUP BY trick that we did above, but based on the current state of each ticket, rather than each event:

      SELECT CONTACT_ID, 
          SUM(CASE WHEN STATUS='new' THEN 1 ELSE 0 END) AS TICKETS_NEW, 
          SUM(CASE WHEN STATUS='processing' THEN 1 ELSE 0 END) AS TICKETS_PROCESSING, 
          SUM(CASE WHEN STATUS='waiting' THEN 1 ELSE 0 END) AS TICKETS_WAITING, 
          SUM(CASE WHEN STATUS='close-request' THEN 1 ELSE 0 END) AS TICKETS_CLOSEREQUEST ,
          SUM(CASE WHEN STATUS='closed' THEN 1 ELSE 0 END) AS TICKETS_CLOSED
      FROM TICKETS_TABLE 
      GROUP BY CONTACT_ID;
    

    This gives us what we want:

      1472 | 0 | 0 | 1 | 1 | 1
    
  9. Let's feed another ticket's events into the topic and observe how the table's state changes. Rows from a table are re-emitted when the state changes; you can also cancel the SELECT and rerun it to see the current state only.

    Sample data to try it for yourself:

    {"ID":8946,"CONTACT_ID":42,"SUBJECT":"","STATUS":"new","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ID":8946,"CONTACT_ID":42,"SUBJECT":"","STATUS":"processing","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ID":8946,"CONTACT_ID":42,"SUBJECT":"","STATUS":"waiting","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ID":8946,"CONTACT_ID":42,"SUBJECT":"","STATUS":"processing","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ID":8946,"CONTACT_ID":42,"SUBJECT":"","STATUS":"waiting","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ID":8946,"CONTACT_ID":42,"SUBJECT":"","STATUS":"closed","TIMESTRING":"2019-04-16 23:07:01.000"}
    {"ID":8946,"CONTACT_ID":42,"SUBJECT":"","STATUS":"close-request","TIMESTRING":"2019-04-16 23:07:01.000"}
    

If you want to try this out further you can generate an stream of additional dummy data with this from Mockaroo, piped through awk to slow it down so you can see the effect on the generated aggregates as each message arrives:

while [ 1 -eq 1 ]
  do curl -s "https://api.mockaroo.com/api/f2d6c8a0?count=1000&key=ff7856d0" | \
      awk '{print $0;system("sleep 2");}' | \
      kafkacat -b localhost -t tickets -P
  done

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.