Блог

Я пытаюсь подключить свою тему кафки к своему BQ DW с помощью разъема приемника kafka BigQuery с открытым исходным кодом. Сообщения, которые я получаю, просты, например:

 {  "test_col1": "value1",  "test_col2": "value2",  "test_col3": "value3" }  

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

 value.converter.schemas.enable: false key.converter.schemas.enable: false  

Я смог преодолеть эту проблему. Однако теперь параметр allowNewBigQueryFields для автоматического создания полей и параметры автосоздания таблиц больше не работают. Я получаю исключение Nullpointerexception при включении этих настроек. Означает ли это, что он ищет схему где-то, либо в сообщении, либо в реестре схем, но не находит ее? Могу ли я включить schemas.enable как false, а также ожидать, что соединитель автоматически создаст таблицу и столбцы? Как у меня может быть тема кафки, которая автоматически создает таблицу и столбцы в BQ, когда мои сообщения являются обычным json? Мои фактические темы намного сложнее и могут часто менять столбцы, поэтому предварительное указание схемы не является для меня хорошим решением.