# INNER JOIN STREAM with a Lookup-Table
**URL:** https://forum.confluent.io/t/inner-join-stream-with-a-lookup-table/4106
**Category:** ksqlDB
**Created:** [7 February 2022 16:59 UTC](https://forum.confluent.io/t/inner-join-stream-with-a-lookup-table/4106 "2022-02-07T16:59:42Z")
**Posts on this page:** 4
**Page:** 1
### Author:  [@KevinBegrow](https://forum.confluent.io/u/KevinBegrow)
#### Post date: [7 February 2022 16:59 UTC](https://forum.confluent.io/t/inner-join-stream-with-a-lookup-table/4106/1 "2022-02-07T16:59:42Z")
Hello Everyone,
I hope someone here can help me with this issue.
I have a lookup table which comes from an kafka topic provided via jdbc connector (using Avro key/value converter and schema registry).
```auto
key:
{"product_key": "0286-04KARTE"}
Value:
{"product_key":"0286-04KARTE","product_id":2888793,"customer_id":10004}
```
Then I get via REST Connector product attributes.
I want to JOIN the attribute STREAM with the product TABLE to enrich the products with those informations.
I do get data into flattend\_product\_attributes correctly but when I do the JOIN I get no results.
I do get warnings like:
```auto
[2022-02-07 16:39:47,715] WARN Skipping record due to null key. topic=[input_products] partition=[6] offset=[400275] (org.apache.kafka.streams.kstream.internals.KTableSource)
```
I suppose this is ok when the key is not existing in Table. But there are also product\_keys which exists in both.
KSQLDB Statements:
```auto
CREATE TABLE products_lookup (
product_id BIGINT,
product_key VARCHAR PRIMARY KEY,
customer_id INT
) WITH (kafka_topic='input_products', partitions=30, key_format='AVRO', value_format='AVRO');
CREATE STREAM product_attributes
( gtin BIGINT,
created VARCHAR,
manufacturerAttributes STRUCT <
colorName VARCHAR,
"size" VARCHAR,
bteCode VARCHAR,
material VARCHAR,
functions VARCHAR,
name VARCHAR,
description VARCHAR,
colorCode VARCHAR,
shortDescription VARCHAR,
category VARCHAR,
manufacturedIn VARCHAR,
customsTariffNumber VARCHAR
>,
media STRUCT <
images ARRAY >>,
fcAttributes STRUCT <
animalParts VARCHAR,
targetGroup VARCHAR,
material VARCHAR,
careInstruction ARRAY,
season VARCHAR,
ageGroup VARCHAR,
category VARCHAR,
searchColors VARCHAR
>,
fcKeys STRUCT <
animalParts VARCHAR,
targetGroup VARCHAR,
materials ARRAY
>,
partName VARCHAR
>
>,
careInstruction ARRAY,
season VARCHAR,
ageGroup VARCHAR,
category VARCHAR,
searchColors VARCHAR,
seasonYear VARCHAR
>,
articleNumber VARCHAR,
permissions STRUCT <
erp VARCHAR,
ecommerce VARCHAR,
digitalWindow VARCHAR
>,
colorGroupId VARCHAR,
styleGroupId VARCHAR,
_id VARCHAR,
brand STRUCT <
id VARCHAR
>,
updated VARCHAR
)
WITH (KAFKA_TOPIC='input_product_attributes',
VALUE_FORMAT='JSON',
partitions=30);
CREATE STREAM flattend_product_attributes
WITH (VALUE_FORMAT='AVRO') AS
SELECT
gtin,
created AS CREATEDAT,
manufacturerAttributes->colorName AS fc_mfa_colorName,
manufacturerAttributes->"size" AS fc_mfa_size,
manufacturerAttributes->bteCode AS fc_mfa_bteCode,
manufacturerAttributes->material AS fc_mfa_material,
manufacturerAttributes->functions AS fc_mfa_functions,
manufacturerAttributes->name AS fc_mfa_name,
manufacturerAttributes->description AS fc_mfa_description,
manufacturerAttributes->colorCode AS fc_mfa_colorCode,
manufacturerAttributes->shortDescription AS fc_mfa_shortDescription,
manufacturerAttributes->category AS fc_mfa_category,
manufacturerAttributes->manufacturedIn AS fc_mfa_manufacturedIn,
manufacturerAttributes->customsTariffNumber AS fc_mfa_customsTariffNumber,
EXPLODE(media->images)->isPreliminary AS fc_media_images_isPreliminary,
EXPLODE(media->images)->isMainImage AS fc_media_images_isMainImage,
EXPLODE(media->images)->shootType AS fc_media_images_shootType,
EXPLODE(media->images)->_id AS fc_media_images_id,
EXPLODE(media->images)->productView AS fc_media_images_productView,
EXPLODE(media->images)->hash AS fc_media_images_hash,
fcAttributes->animalParts AS fc_attr_animalParts,
fcAttributes->targetGroup AS fc_attr_targetGroup,
fcAttributes->material AS fc_attr_material,
EXPLODE(fcAttributes->careInstruction) AS fc_attr_careInstruction,
fcAttributes->season AS fc_attr_season,
fcAttributes->ageGroup AS fc_attr_ageGroup,
fcAttributes->category AS fc_attr_category,
fcAttributes->searchColors AS fc_attr_searchColors,
articleNumber AS product_key,
colorGroupId AS fc_colorGroupId,
styleGroupId AS fc_styleGroupId,
_id AS fc_id,
brand->id AS fc_brand_id,
updated AS CHANGEDAT
FROM product_attributes
PARTITION BY articleNumber;
CREATE STREAM product_attributes_flatten_data_combined
WITH (VALUE_FORMAT='AVRO' ) AS
SELECT
p.product_id,
a.product_key,
AS_VALUE(a.product_key) AS parent_product_key,
a.fc_mfa_colorName,
a.fc_mfa_bteCode,
a.fc_mfa_material,
a.fc_mfa_name,
a.fc_mfa_colorCode,
a.fc_mfa_description,
a.fc_mfa_shortDescription,
a.fc_mfa_category,
a.fc_mfa_manufacturedIn,
a.fc_mfa_customsTariffNumber,
a.fc_colorGroupId,
a.fc_styleGroupId,
a.fc_brand_id,
p.customer_id,
a.CREATEDAT,
a.CHANGEDAT
FROM flattend_product_attributes a
INNER JOIN products_lookup p ON a.product_key = p.product_key;
```
---
### Author:  [@KevinBegrow](https://forum.confluent.io/u/KevinBegrow)
#### Post date: [7 February 2022 17:38 UTC](https://forum.confluent.io/t/inner-join-stream-with-a-lookup-table/4106/2 "2022-02-07T17:38:45Z")
Its it connected with this?
>
>
> KSQL supports joining streams to tables. However, for this to work, the \*\*table'…s underlying kafka topic\*\* must have as a \*\*key\*\* the \*\*column on which the join is made\*\*. Currently KSQL \_silently\_ fails to make a join in which it is non-obvious to the user (particularly one from the Database world and familiar with SQL) why it doesn't work.
>
> Consider a simple stream/table (event/reference, a.k.a. fact/dimension) join:
>
> \* \`RENTAL\` is a stream of rental events, with various foreign key relationships including a \`CUSTOMER\_ID\`
> \* \`CUSTOMER\` is a table of customer information, with a primary key of \`CUSTOMER\_ID\`
>
> The data in this example comes from MySQL, connected into Kafka using Debezium.
>
> \*\*MySQL\*\*:
>
> \* The event data:
>
> mysql\> SELECT R.RENTAL\_ID, R.RENTAL\_DATE, R.CUSTOMER\_ID FROM RENTAL R WHERE R.CUSTOMER\_ID=603;
> +-----------+---------------------+-------------+
> | RENTAL\_ID | RENTAL\_DATE | CUSTOMER\_ID |
> +-----------+---------------------+-------------+
> | 16050 | 2018-02-16 18:31:55 | 603 |
> | 16051 | 2018-02-16 18:34:37 | 603 |
> | 16052 | 2018-02-16 18:35:01 | 603 |
> | 16053 | 2018-02-16 18:35:07 | 603 |
> +-----------+---------------------+-------------+
> 4 rows in set (0.00 sec)
>
> \* The reference data:
>
> mysql\> SELECT C.FIRST\_NAME,C.LAST\_NAME,C.CUSTOMER\_ID FROM CUSTOMER C WHERE C.CUSTOMER\_ID=603;
> +------------+-----------+-------------+
> | FIRST\_NAME | LAST\_NAME | CUSTOMER\_ID |
> +------------+-----------+-------------+
> | BOB | Astley | 603 |
> +------------+-----------+-------------+
> 1 row in set (0.00 sec)
>
> \* The executed join in MySQL:
>
> mysql\> SELECT R.RENTAL\_ID, R.RENTAL\_DATE, R.CUSTOMER\_ID, C.FIRST\_NAME, C.LAST\_NAME FROM RENTAL R LEFT OUTER JOIN CUSTOMER C ON R.CUSTOMER\_ID=C.CUSTOMER\_ID WHERE C.FIRST\_NAME IS NOT NULL AND R.CUSTOMER\_ID=603;
> +-----------+---------------------+-------------+------------+-----------+
> | Rental\_ID | RENTAL\_DATE | CUSTOMER\_ID | FIRST\_NAME | LAST\_NAME |
> +-----------+---------------------+-------------+------------+-----------+
> | 16050 | 2018-02-16 18:31:55 | 603 | BOB | Astley |
> | 16051 | 2018-02-16 18:34:37 | 603 | BOB | Astley |
> | 16052 | 2018-02-16 18:35:01 | 603 | BOB | Astley |
> | 16053 | 2018-02-16 18:35:07 | 603 | BOB | Astley |
> +-----------+---------------------+-------------+------------+-----------+
> 4 rows in set (0.00 sec)
>
> \*\*Now the same in KSQL\*\*:
>
> \* Stream:
>
> ksql\> CREATE STREAM RENTAL WITH (KAFKA\_TOPIC='fullfillment.sakila.rental-flat', VALUE\_FORMAT='AVRO');
>
> Message
> ----------------
> Stream created
> ----------------
>
> ksql\> SELECT R.RENTAL\_ID, R.RENTAL\_DATE, R.CUSTOMER\_ID FROM RENTAL R WHERE R.CUSTOMER\_ID=603;
> 16050 | 1518805915000 | 603
> 16051 | 1518806077000 | 603
> 16052 | 1518806101000 | 603
> 16053 | 1518806107000 | 603
>
> \* Table:
>
> ksql\> CREATE TABLE CUSTOMER WITH (KAFKA\_TOPIC='fullfillment.sakila.customer-flat',VALUE\_FORMAT='AVRO',KEY='customer\_id');
>
> Message
> ---------------
> Table created
> ---------------
>
> ksql\> SELECT C.FIRST\_NAME,C.LAST\_NAME,C.CUSTOMER\_ID FROM CUSTOMER C WHERE C.CUSTOMER\_ID=603;
> BOB | Astley | 603
>
> \* Join:
>
> ksql\> SELECT R.RENTAL\_ID, R.RENTAL\_DATE, R.CUSTOMER\_ID, C.FIRST\_NAME, C.LAST\_NAME FROM RENTAL R LEFT OUTER JOIN CUSTOMER C ON R.CUSTOMER\_ID=C.CUSTOMER\_ID WHERE C.FIRST\_NAME IS NOT NULL AND R.CUSTOMER\_ID=603;
> \[... no output ...\]
>
> \---
>
> Here is the problem. The key that we declared for the table (\`KEY='customer\_id'\`) \_does not match the key for the Kafka message\_:
>
> ksql\> SELECT ROWKEY,CUSTOMER\_ID FROM CUSTOMER WHERE CUSTOMER\_ID=603;
> � | 603
>
> Examining the underlying Kafka topic:
>
> $ kafkacat -C -K: -b localhost:9092 -f 'Key: %k\\nKey Bytes: %K\\nValue: %s\\nValue Bytes: %S\\n\\n' -t fullfillment.sakila.customer-flat
>
> Key: �
> Key Bytes: 7
> Value: Y� BOBAstley����X22018-02-16T18:47:58+01:00
> Value Bytes: 54
>
> Same data, Avro deserialised:
>
> $ ./bin/kafka-avro-console-consumer \\
> --bootstrap-server localhost:9092 \\
> --property schema.registry.url=http://localhost:8081 \\
> --property print.key=true \\
> --topic fullfillment.sakila.customer-flat \\
> --from-beginning
>
> {"customer\_id":603} {"customer\_id":603,"store\_id":1,"first\_name":"BOB","last\_name":"Astley","email":null,"address\_id":1,"active":1,"create\_date":1518805799000,"last\_update":"2018-02-16T18:47:58+01:00"}
>
> \---
>
> So \*technically\* KSQL is evaluating the join correctly, but \_in practice\_ this is going to suck for the end user, particularly one who is not familiar with Kafka's key/value message structure.
>
> The workaround is to manually rekey the topic:
>
> ksql\> CREATE STREAM CUST\_RAW\_STREAM WITH (KAFKA\_TOPIC='fullfillment.sakila.customer-flat', VALUE\_FORMAT='AVRO');
>
> Message
> ----------------
> Stream created
> ----------------
>
> ksql\> CREATE STREAM CUSTOMER\_REKEYED AS SELECT \* FROM CUST\_RAW\_STREAM PARTITION BY CUSTOMER\_ID;
>
> Message
> ----------------------------
> Stream created and running
> ----------------------------
>
> ksql\> CREATE TABLE customer WITH (KAFKA\_TOPIC='CUSTOMER\_REKEYED', VALUE\_FORMAT='avro', KEY='CUSTOMER\_ID');
>
> Message
> ---------------
> Table created
> ---------------
>
> The resulting topic is keyed correctly (i.e. the key \_is\_ the \`CUSTOMER\_ID\`):
>
> $ kafkacat -C -K: -b localhost:9092 -f 'Key: %k\\nKey Bytes: %K\\nValue: %s\\nValue Bytes: %S\\n\\n' -t CUSTOMER\_REKEYED
> Key: 603
> Key Bytes: 3
> Value: ~� BOBAstley����X22018-02-16T18:47:58+01:00
> Value Bytes: 62
>
> Now in the KSQL table the ROWKEY matches CUSTOMER\_ID:
>
> ksql\> SELECT ROWKEY,CUSTOMER\_ID FROM CUSTOMER WHERE CUSTOMER\_ID=603;
> 603 | 603
>
> and the desired join succeeds:
>
> ksql\> SELECT R.RENTAL\_ID, R.RENTAL\_DATE, R.CUSTOMER\_ID, C.FIRST\_NAME, C.LAST\_NAME FROM RENTAL R LEFT OUTER JOIN CUSTOMER C ON R.CUSTOMER\_ID=C.CUSTOMER\_ID WHERE C.FIRST\_NAME IS NOT NULL;
> 16050 | 1518805915000 | 603 | BOB | Astley
> 16051 | 1518806077000 | 603 | BOB | Astley
> 16052 | 1518806101000 | 603 | BOB | Astley
> 16053 | 1518806107000 | 603 | BOB | Astley
>
> How do we make this less painful for the user? Several ideas:
>
> 1. Some kind of rekey operation that takes a \`CREATE TABLE\` declaration and explicitly rekeys the source topic (i.e. implements the above workaround).
> This could be done:
> \* automatically (best UX for new users, but overkill if already correctly keyed)
> \* with performance optimisation to automagically skip the rekey if the topic was already keyed on the declared table key.
> \* and/or with explicit '\`NOREKEY\`' syntax
> \* on demand e.g. with a \`REKEY\` option in the \`CREATE TABLE\` declaration (requires users to know to look for the option)
>
> 2. Evaluate a sample of messages and warn the user if the message key doesn't match the declared table key
>
> Something that would also help reduce instances of this -but not avoid the problem entirely- would be to support different Key formats (in this case, the Key \_is\_ the declared \`CUSTOMER\_ID\`, but is serialised as Avro not String that KSQL currently assumes)
>
> \---
>
> Interestingly, a side-effect of KSQL only using \`STRING\` Keys is that the message on the derived topic cannot be read using avro-console-consumer if \`print.key=true\`:
>
> \`\`\`
> Robin@asgard02 ~/c/confluent-4.0.0\> ./bin/kafka-avro-console-consumer \\
> --bootstrap-server localhost:9092 \\
> --property schema.registry.url=http://localhost:8081 \\
> --property print.key=true --topic CUSTOMER\_REKEYED --from-beginning
> Processed a total of 1 messages
> \[2018-02-16 19:11:53,976\] ERROR Unknown error when running consumer: (kafka.tools.ConsoleConsumer$:107)
> org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
> Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
> \`\`\`
Because the underlying Kafka Topic of product\_attributes doesn not have a message key set.
---
### Author:  [@KevinBegrow](https://forum.confluent.io/u/KevinBegrow)
#### Post date: [8 February 2022 08:35 UTC](https://forum.confluent.io/t/inner-join-stream-with-a-lookup-table/4106/3 "2022-02-08T08:35:40Z")
thanks to this workaround I got it fixed.
>
>
> KSQL currently requires a topic backing a Table to have message keys equal to th…e declared key of the table. If they are not, then the Table will not function as intended, with no error or warning. This causes confusion and frustration for the user:
>
> \* https://github.com/confluentinc/ksql/issues/749
> \* https://github.com/confluentinc/ksql/issues/1405
>
> This is a frequent occurance as keys on Kafka messages are not mandatory and may well not be set on a source topic. In addition, the target user for KSQL may not initially be familiar with the concept of keys in Kafka messages and so overlook this requirement entirely.
>
> Take an example in which it is desired to join a stream of events \`wibble\` to a table \`FOO\` of lookup details, on the common join column \`BAR\`. The source \`foo\` topic is not keyed. Done like this, the join would fail silently:
>
> CREATE TABLE foo\_table \\
> WITH (KAFKA\_TOPIC='foo', \\
> VALUE\_FORMAT='AVRO', \\
> KEY='BAR');
>
> SELECT \* FROM wibble w INNER JOIN foo\_table f on w.BAR=f.BAR;
>
> The join would fail silently because even though a \`KEY\` is declared in the \`CREATE TABLE\`, it does not match \_the actual key of the Kafka topic\_. This is the case for both null-keyed messages, as well as messages with a key \*but a key that does not match the value of the \`BAR\` column in the message value\*.
>
> The current workaround is:
>
> 1. Register the source topic as a stream:
>
> CREATE STREAM foo\_stream \\
> WITH (KAFKA\_TOPIC='foo', \\
> VALUE\_FORMAT='AVRO');
>
> 2. Repartition the topic on the desired key, writing to a new one:
>
> CREATE STREAM FOO\_REKEYED AS \\
> SELECT \* FROM foo \\
> PARTITION BY BAR;
>
> Note that by repartitioning, all ordering guarentees are lost.
>
> 3. Declare a table over the \_rekeyed topic\_:
>
> CREATE TABLE foo\_table2 \\
> WITH (KAFKA\_TOPIC='FOO\_REKEYED', \\
> VALUE\_FORMAT='AVRO', \\
> KEY='BAR');
>
> Of these three steps, two are redundant. It should only be necessary to declare the \`TABLE\` and its key, and KSQL handle the rest. This is particularly true given that in the DDL \`WITH\` clause, \`KEY\` is specified which would leave most users to understand that KSQL will act on it, or at the very least throw an error if the message key does not match the declared one.
>
> Consideration would need to be given to the issue of order guarentees, so perhaps KSQL should have a DDL option to force the rekey, or to throw an error if the declared key doesn't match the message key.
>
> Related discussion:
>
> \* https://github.com/confluentinc/ksql/issues/804
Thank you @rmoff
---
### Author:  [@system](https://forum.confluent.io/u/system)
#### Post date: [15 February 2022 08:36 UTC](https://forum.confluent.io/t/inner-join-stream-with-a-lookup-table/4106/4 "2022-02-15T08:36:07Z")
This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.