Как я могу получить данные JSON из Kafka с помощью SparkStreaming

Question

Я использую kafka monitor для изменения LocalFile и SparkStreaming для анализа.Но я не могу извлечь данные из kafka, потому что формат данных - JSON.

When I tap the command bin/kafka-console-consumer.sh --bootstrap-server master:9092,slave1:9092,slave2:9092  --topic kafka-streaming --from-beginning, 

ФОРМАТ ДАННЫХ:

{
  "schema": {
    "type": "string",
    "optional": false
  },
  "payload": "{\"like_count\": 594, \"view_count\": 49613, \"user_name\": \"   w\", \"play_url\": \"http://upic/2019/04/08/12/BMjAxOTA0MDgxMjQ4MTlfMjA3ODc2NTY2XzEyMDQzOTQ0MTc4XzJfMw==_b_Bfa330c5ca9009708aaff0167516a412d.mp4?tag=1-1555248600-h-0-gjcfcmzmef-954d5652f100c12e\", \"description\": \"ţ ų   ඣ    9 9 9 9\", \"cover\": \"http://uhead/AB/2016/03/09/18/BMjAxNjAzMDkxODI1MzNfMjA3ODc2NTY2XzJfaGQ5OQ==.jpg\", \"video_id\": 5235997527237673952, \"comment_count\": 39, \"download_url\": \"http://2019/04/08/12/BMjAxOTA0MDgxMjQ4MTlfMjA3ODc2NTY2XzEyMDQzOTQ0MTc4XzJfMw==_b_Bfa330c5ca9009708aaff0167516a412d.mp4?tag=1-1555248600-h-1-zdpjkouqke-5862405191e4c1e4\", \"user_id\": 207876566, \"video_create_time\": \"2019-04-08 12:48:21\", \"user_sex\": \"F\"}"
}

Версия spark - 2.3.0 а версия кафки 1.1.0.Версия spark-streaming-kafka - 0-10_2.11-2.3.0.

Данные JSON в столбце PAYOAD - это то, с чем я хочу разобраться и проанализировать.Как я могу изменить коды для получения данных JSON

Answer 1

Используйте org.apache.kafka.common.serialization.StringDeserializer и org.apache.kafka.common.serialization.StringSerializer для потребления и отправки данных в тему кафки соответственно. Таким образом, вы получите String для потребления, которую очень легко конвертировать в JSON Object using JSONParser