利用kafka监控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,kafka的版本是1.1.0。spark streaming kafka的版本是0-10μ2.11-2.3.0。
payad列中的json数据是我要处理和分析的。如何更改代码以获取json数据
1条答案
按热度按时间vsnjm48y1#
使用
org.apache.kafka.common.serialization.StringDeserializer
以及org.apache.kafka.common.serialization.StringSerializer
分别用于消费和向Kafka主题发送数据。通过这种方式,您将在消费时获得一个字符串,该字符串可以非常容易地转换为json对象using JSONParser