INNER JOIN STREAM with a Lookup-Table

KevinBegrow · 7 February 2022 16:59

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).

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:

[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:

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 <STRUCT <
                            isPreliminary VARCHAR,
                            isMainImage VARCHAR,
                            shootType VARCHAR,
                            _id VARCHAR,
                            productView VARCHAR,
                            hash VARCHAR
                        >>>,
                      fcAttributes STRUCT <
                        animalParts VARCHAR,
                        targetGroup VARCHAR,
                        material VARCHAR,
                        careInstruction ARRAY<VARCHAR>,
                        season VARCHAR,
                        ageGroup VARCHAR,
                        category VARCHAR,
                        searchColors VARCHAR
                      >,
                      fcKeys STRUCT <
                        animalParts VARCHAR,
                        targetGroup VARCHAR,
                        materials ARRAY <STRUCT <
                            composition ARRAY < STRUCT <
                                material VARCHAR,
                                percentage INT
                              >
                            >,
                            partName VARCHAR
                            >
                          >,
                        careInstruction ARRAY<VARCHAR>,
                        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;

KevinBegrow · 7 February 2022 17:38

Its it connected with this?

github.com/confluentinc/ksql

KSQL joins will fail silently if a table's topic message key does not match the declared KSQL Table key

opened 07:16PM - 16 Feb 18 UTC

rmoff

bug record-keys user-experience join-functions streaming-engine

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.

KevinBegrow · 8 February 2022 08:35

thanks to this workaround I got it fixed.

github.com/confluentinc/ksql

Support repartitioning of topic data for stream-table joins more elegantly

opened 01:54PM - 03 Jan 19 UTC

rmoff

data-accessibility record-keys user-experience

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

system · 15 February 2022 08:36

This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.

Topic		Replies	Views
KsqlDB Table-Table join on STRUCT Type Primary Key ksqlDB	1	512	29 February 2024
Join Stream-Table (ORDER-PROVIDER) when the stream does not have the PROVIDER'S ID but others informations ksqlDB	8	4305	26 July 2021
Trouble with STREAM STREAM JOIN ksqlDB	4	3342	22 April 2021
Missing records in STREAM-TABLE join ksqlDB	6	1683	16 September 2023
Create table as select produces null values ksqlDB	4	3571	20 May 2021

INNER JOIN STREAM with a Lookup-Table

Related Topics