confluentinc / confluentinc/kafka-tutorials
Working with grouped messages
- Dominant language
- Java
- Stars
- 39
- Forks
- 91
- PR merge metrics
- No merged PRs in 30d
Description
**Describe the use case**
We want to keep track of the current address for some persons. We get updates on their location per country, but these updates contain all the persons moved into that country that 'day'. So we need to break those down in single messages using `EXPLODE`, before we can create the `JOIN` for the updates.
**Provide the ksqlDB application**
Document the full set of ksqlDB statements that represent the use case.
[statements as exucuted](https://github.com/gklijs/ksqlDB-GraphQL-poc/blob/main/rust-data-creator/src/ksqldb.rs)
```
CREATE TABLE IF NOT EXISTS persons (id_key STRING PRIMARY KEY) WITH (KAFKA_TOPIC = 'persons', VALUE_FORMAT = 'PROTOBUF');
CREATE STREAM IF NOT EXISTS address_updates (country_id STRING KEY) WITH (KAFKA_TOPIC = 'address-updates', VALUE_FORMAT = 'PROTOBUF');
CREATE STREAM IF NOT EXISTS exploded_address_updates AS SELECT country_id, country, EXPLODE(persons) AS person_id, EXPLODE(addresses)
AS address FROM address_updates;
CREATE STREAM IF NOT EXISTS persons_with_address AS SELECT persons.id_key AS id, persons.first_name as first_name, persons.last_name
as last_name, persons.birthday as birthday, country, address FROM exploded_address_updates JOIN persons ON
exploded_address_updates.person_id = persons.id_key;
"#;
```
Contributor guide
Assessment
This issue has not been assessed yet.