如何使用 Spark Streaming 从 Kafka 获取 JSON 数据答案

【问题标题】：How can I acqurie the JSON data from Kafka using SparkStreaming如何使用 Spark Streaming 从 Kafka 获取 JSON 数据
【发布时间】：2019-04-15 15:35:51
【问题描述】：

我使用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。

PAYOAD 列中的 JSON 数据是我要处理和分析的。如何更改代码以获取 JSON 数据

【问题讨论】：

标签： scala apache-kafka

【解决方案1】：

使用org.apache.kafka.common.serialization.StringDeserializer 和org.apache.kafka.common.serialization.StringSerializer 分别消费和发送数据到kafka 主题。这样，您将获得一个消费字符串，它可以很容易地转换为 JSON 对象using JSONParser

【讨论】：

谢谢！我解决问题！在“connect-standalone.properties”文件中，“key.converter.schemas.enable”和“value.converter.schemas.enable”为“false”，“schema”的数据被擦除。接下来我使用stream.map(record => record.replace()) 用于擦除“PAYLOAD”数据中的“\”和“”