Revision history [back]

You can include group by keys in the message body using ksqldb by using the GROUP BY clause and the SELECT statement. The group by key can be added as an additional field to the message body by selecting it in the SELECT statement.

For example, let's say you have a stream of events with fields user_id, action, and timestamp, and you want to group the events by user_id:

CREATE STREAM events (
  user_id VARCHAR,
  action VARCHAR,
  timestamp BIGINT
) WITH (
  kafka_topic='events_topic',
  value_format='JSON'
);

SELECT user_id, action, COUNT(*) as count
FROM events
GROUP BY user_id
EMIT CHANGES;

In the above example, we are grouping the events by user_id and counting the number of events for each user. The user_id field is included in the final message body as a group by key.

You can also add additional fields to the message body by selecting them in the SELECT statement. For example:

SELECT user_id, action, COUNT(*) as count, MAX(timestamp) as latest_timestamp
FROM events
GROUP BY user_id
EMIT CHANGES;

In the above example, we are selecting the action, count, and the max value of the timestamp field for each user. The final message body includes all of these fields, as well as the user_id group by key.