Блог

Я довольно новичок в spark streaming и scala. У меня есть данные Json и некоторые другие случайные данные журнала, поступающие из темы kafka.Я смог отфильтровать только данные json следующим образом

 val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet).map(_._2).filter (x => x.matches("^[{].*" ))
 

Мои данные json выглядят следующим образом.

 {"time":"125573","randomcol":"abchdre","address":{"city":"somecity","zip":"123456"}}
 

Я пытаюсь проанализировать данные json и поместить их в таблицу hive.
может кто-нибудь, пожалуйста, указать мне правильное направление.
Спасибо

Для этого есть несколько способов.

1. создайте внешнюю таблицу hive с необходимыми столбцами и указанием на это местоположение данных.
2. При создании таблицы вы можете использовать стандартный JSON serde, а затем использовать функцию get_json_object hive и загрузить эти необработанные данные в конечную таблицу. Обратитесь к этому для получения подробной информации о функции

или

Вы можете попробовать avro serde и указать схему avro в соответствии с вашим сообщением json для создания таблицы hive. Обратитесь к этому для примера avro serde .

Надеюсь, это поможет