#apache-kafka #left-join #confluent-platform #ksqldb
Вопрос:
Я изучаю K-SQL/KSQL-DB и в настоящее время изучаю соединения. Ниже приведена проблема, в которой я застрял.
У меня есть 1 поток «DRIVERSTREAMREPARTITIONEDKEYED» и одна таблица «СТРАНЫ», ниже их описание.
ksql> describe DRIVERSTREAMREPARTITIONEDKEYED;
Name: DRIVERSTREAMREPARTITIONEDKEYED
Field | Type
--------------------------------------
COUNTRYCODE | VARCHAR(STRING) (key)
NAME | VARCHAR(STRING)
RATING | DOUBLE
--------------------------------------
ksql> describe countries;
Name : COUNTRIES
Field | Type
----------------------------------------------
COUNTRYCODE | VARCHAR(STRING) (primary key)
COUNTRYNAME | VARCHAR(STRING)
----------------------------------------------
Это примерные данные, которые у них есть,
ksql> select * from DRIVERSTREAMREPARTITIONEDKEYED emit changes;
--------------------------------------------- --------------------------------------------- ---------------------------------------------
|COUNTRYCODE |NAME |RATING |
--------------------------------------------- --------------------------------------------- ---------------------------------------------
|SGP |Suresh |3.5 |
|IND |Mahesh |2.4 |
ksql> select * from countries emit changes;
--------------------------------------------------------------------- ---------------------------------------------------------------------
|COUNTRYCODE |COUNTRYNAME |
--------------------------------------------------------------------- ---------------------------------------------------------------------
|IND |INDIA |
|SGP |SINGAPORE |
Я пытаюсь выполнить «левое внешнее» соединение с ними, при этом поток находится с левой стороны, но ниже приведен результат, который я получаю,
select d.name,d.rating,c.COUNTRYNAME from DRIVERSTREAMREPARTITIONEDKEYED d left join countries c on d.COUNTRYCODE=c.COUNTRYCODE emit changes;
--------------------------------------------- --------------------------------------------- ---------------------------------------------
|NAME |RATING |COUNTRYNAME |
--------------------------------------------- --------------------------------------------- ---------------------------------------------
|Suresh |3.5 |null |
|Mahesh |2.4 |null |
В идеальном сценарии я должен получить данные в столбце «ИМЯ СТРАНЫ», так как столбец «КОД СТРАНЫ» как в потоке, так и в данных содержит соответствующие данные.
Я много пытался искать, но безрезультатно. Я использую «Платформу слияния: 6.1.1»
Ответ №1:
Для работы соединения мы несем ответственность за проверку того, находятся ли ключи обеих присоединяемых сущностей в одном разделе, KsqlDB не может проверить, одинаковы ли стратегии разделения для обоих входных данных соединения.
В моем случае в моей теме «Драйверы» было 2 раздела, на которых я создал поток «Поток драйверов», в котором, в свою очередь, также было 2 раздела, но в таблице «Страны», к которой я хотел присоединиться, был только 1 раздел, из-за этого я «перенастроил «»Поток драйверов» и создал другой поток «Поток драйверов», показанный в вопросе.
Но данные таблицы и потока не находились в одном разделе, поэтому соединение не удалось.
Я создал еще одну тему с 1 разделом «DRIVERINFO».
kafka-topics --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic DRIVERINFO
Затем создал над ним поток «DRIVERINFOSTREAM».
CREATE STREAM DRIVERINFOSTREAM (NAME STRING, RATING DOUBLE, COUNTRYCODE STRING) WITH (KAFKA_TOPIC='DRIVERINFO', VALUE_FORMAT='JSON');
Наконец присоединился к нему с таблицей «СТРАНЫ», которая, наконец, сработала.
ksql> select d.name,d.rating,c.COUNTRYNAME from DRIVERINFOSTREAM d left join countries c on d.COUNTRYCODE=c.COUNTRYCODE EMIT CHANGES;
------------------------------------------- ------------------------------------------- -------------------------------------------
|NAME |RATING |COUNTRYNAME |
------------------------------------------- ------------------------------------------- -------------------------------------------
|Suresh |2.4 |SINGAPORE |
|Mahesh |3.6 |INDIA |
Для получения подробной информации перейдите по ссылкам ниже,