Возможный вариант для PrimayKey в создании таблицы с KSQL? - PullRequest
Новая
       16

Возможный вариант для PrimayKey в создании таблицы с KSQL?

0 голосов
/ 14 марта 2019

Я начал работать с KSQL и вполне пережил опыт.Я пытаюсь работать с таблицей и потоком объединения и сценарий, как показано ниже.У меня есть пример данных, подобный этому: 

"0117440512","0134217727","US","United States","VIRGINIA","Vienna","DoD Network Information Center"
"0134217728","0150994943","US","United States","MASSACHUSETTS","Woburn","Genuity"

в моей теме кафки-1.Статический набор данных загружен в таблицу и может обновляться раз в месяц или около того.

У меня есть еще один набор данных, например: 

{"state":"AD","id":"020","city":"Andorra","port":"02","region":"Canillo"}
{"state":"GD","id":"024","city":"Arab","port":"29","region":"Ordino"}

в kafka topic-2.Поток данных загружается в потоки.

Поскольку Table невозможно создать без указания ключа, у моих данных нет уникального столбца для этого.Итак, при загрузке данных из темы-1 в таблицу каким должен быть мой ключ?Помните, что моя таблица может быть заполнена / обновлена ​​один раз в месяц или около того с теми же данными и новые также один раз.После загрузки новых данных я могу заменить их ключом.

Я пытался найти, есть ли что-то вроде инкрементного значения, как мы называем PrimaryKey в SQL, но не нашел ничего.

Может ли кто-нибудь помочь мне исправить мой подход к реализации или запрос на создание PrimaryKey, если он существует.Спасибо

1 Ответ

0 голосов
/ 14 марта 2019

Нет, в KSQL отсутствует концепция самовращающегося ключа.Вы должны определить ключ при создании данных в теме, по которой определяется таблица KSQL.

--- EDIT

Если вы хотите установить ключ для сообщения при его поступлении через Kafka Connect, вы можете использовать Преобразование одного сообщения (SMT).

            "transforms":"createKey,extractInt",
            "transforms.createKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
            "transforms.createKey.fields":"id",
            "transforms.extractInt.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
            "transforms.extractInt.field":"id"

Подробнее см. Здесь .

...