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}