Поток Kafka SQL (KSQL) не работает для данных JSON с вложенными полями - PullRequest
Новая
       12

Поток Kafka SQL (KSQL) не работает для данных JSON с вложенными полями

0 голосов
/ 25 апреля 2018

Я пытаюсь создать поток Kafka в KSQL поверх темы Kafka.У меня есть записи JSON, как показано ниже, в теме Kafka.

{ "venue": { "venue_name": "HATCH", "lon": -71.18291, "lat": 42.36667, "venue_id": 22491322 }, "visibility": "public", "response": "yes", "guests": 0, "member": { "member_id": 237655942, "member_name": "Nts" }, "rsvp_id": 1724941595, "mtime": 1524620970613, "event": { "event_name": "Intro to Soldering", "event_id": "250106100", "time": 1526853600000, "event_url": "https:\/\/www.meetup.com\/Makers-of-HATCH-Makerspace\/events\/250106100\/" }, "group": { "group_topics": [ { "urlkey": "quilting", "topic_name": "Quilting" }, { "urlkey": "robotics", "topic_name": "Robotics" }, { "urlkey": "sewing", "topic_name": "Sewing" }, { "urlkey": "edtech", "topic_name": "Education & Technology" }, { "urlkey": "craftswap", "topic_name": "Crafts" }, { "urlkey": "diy", "topic_name": "DIY (Do It Yourself)" }, { "urlkey": "hacking", "topic_name": "Hacking" }, { "urlkey": "3d-modeling", "topic_name": "3D Modeling" }, { "urlkey": "tools", "topic_name": "Tools" }, { "urlkey": "arduino", "topic_name": "Arduino" }, { "urlkey": "makers", "topic_name": "Makers" }, { "urlkey": "makerspaces", "topic_name": "Makerspaces" }, { "urlkey": "3d-printing", "topic_name": "3D Printing" }, { "urlkey": "laser-cutting", "topic_name": "Laser Cutting" }, { "urlkey": "scrapbook-die-cutting-machines", "topic_name": "Scrapbook die cutting machines." } ], "group_city": "Watertown", "group_country": "us", "group_id": 18457932, "group_name": "Makers of HATCH Makerspace", "group_lon": -71.18, "group_urlname": "Makers-of-HATCH-Makerspace", "group_state": "MA", "group_lat": 42.37 } }

Эти данные были загружены в тему Kafka.

Я создал поток в KSQLкак показано ниже. 

CREATE STREAM meetup_rsvp_raw 
(  Venue varchar, 
   Visibility varchar, 
   Response varchar, 
   Guests integer, 
   Member varchar, 
   rsvp_id bigint, 
   mtime bigint, 
   event varchar, 
   group_info varchar 
) WITH (KAFKA_TOPIC='meetup-rsvp', VALUE_FORMAT='JSON');

Я вижу ноль в поле group_info (последнее поле в потоке kafka).Примечание: Кафка не разрешает мне создавать поле с именем «группа», так как оно является ключевым словом.Поле названо так: group_info. 

ksql> select * from meetup_rsvp_raw limit 2;

1524624181126 | null | {"venue_name":"Houghton's Pond - Blue Hills","lon":-71.09453,"lat":42.208187,"venue_id":1506300} | public | yes | 0 | {"member_id":159617162,"photo":"https://secure.meetupstatic.com/photos/member/7/2/b/c/thumb_215729372.jpeg","member_name":"Tena Kerns"} | 1724949934 | 1524623875376 | {"event_name":"Blue Hills Buck Hill - Easy Pace / Moderate hike","event_id":"250084062","time":1525010400000,"event_url":"https://www.meetup.com/HikeBikeSocialClub/events/250084062/"} | null

1524624181126 | null | {"venue_name":"Community Wholeness Centre CWC","lon":-79.69191,"lat":44.38976,"venue_id":19966962} | public | no | 0 | {"member_id":222279178,"photo":"https://secure.meetupstatic.com/photos/member/d/3/f/c/thumb_273714268.jpeg","member_name":"Natalie Roy"} | 1724949935 | 1524623875430 | {"event_name":"Karate Class - Ken Shin Budo Kai","event_id":"kbsjtmyxgbnc","time":1525129200000,"event_url":"https://www.meetup.com/CWCBarrie/events/250120204/"} | null

Не уверен, что я делаю неправильно, но любые предложения приветствуются.

1 Ответ

0 голосов
/ 26 апреля 2018

Вы правы, ключевое слово "GROUP" в KSQL. Вы не можете переименовать имя поля в операторе CREATE STREAM, так как KSQL не знает, что ваш столбец group_info ссылается на поле group.

Вы можете использовать кавычки вокруг ваших столбцов, чтобы позволить вам импортировать тему (в настоящее время идентификатор в кавычках должен быть в верхнем регистре, но это ошибка), например, 

CREATE STREAM meetup_rsvp_raw 
(  venue varchar, 
   visibility varchar, 
   response varchar, 
   guests integer, 
   member varchar, 
   rsvp_id bigint, 
   mtime bigint, 
   event varchar, 
   "GROUP" varchar 
) WITH (KAFKA_TOPIC='meetup-rsvp', VALUE_FORMAT='JSON');

Обратите внимание, что вам также нужно использовать кавычки при выборе этого поля: 

SELECT `GROUP` from meetup_rsvp_raw limit 5;

Я создал проблему Github для отслеживания отсутствия документации в этой области .

Дайте нам знать, как вы справляетесь с этим.

Спасибо

Andy

...