Spark 1.6下Spark Streaming整合Twitter JSON转DataFrame遇_corrupt_record问题

我完全懂你现在的困扰——在Spark 1.6里对接Twitter流数据时，想把完整的JSON字段都转成DataFrame，结果所有内容全跑到_corrupt_record列里了。这个问题其实是数据格式传递错了，咱们一步步来解决：

问题根源 #

你当前代码里用json.loads(rawTweet[1])把Kafka传来的JSON字符串解析成了Python字典，但Spark 1.6的sqlContext.read.json() API只接受每行是原始JSON字符串的RDD，而不是已经解析好的字典对象。当它拿到字典时，会把整个对象当成不符合JSON格式的内容，直接扔进损坏记录列。

另外你示例里的{u'quote_count': ...}是Python字典的字符串表示（用单引号），这不是标准JSON格式（标准JSON要求双引号），这也会加剧Spark的解析失败。

解决方案 #

方案1：直接传递原始JSON字符串（推荐） #

去掉json.loads()这一步，让RDD保留原始的JSON字符串，直接交给Spark解析：

def process(time, rdd):
    print("========= %s =========" % str(time))
    try:
        sqlContext = getSqlContextInstance(rdd.context)
        # 直接传入原始JSON字符串组成的RDD
        jsonRDD = sqlContext.read.json(rdd)
        jsonRDD.registerTempTable("tweets")
        jsonRDD.printSchema()
        jsonRDD.show(5)
    except Exception as e:
        print(f"处理出错: {str(e)}")
        pass

rawKafkaStream = KafkaUtils.createStream(ssc, zkQuorum, "kafka-consumer", {kafkaTopic: 4})
# 直接取Kafka消息的value部分，保留原始JSON字符串
parsed_stream = rawKafkaStream.map(lambda rawTweet: rawTweet[1])
parsed_stream.foreachRDD(process)

方案2：如果必须先解析字典再处理 #

如果你需要先对字典做自定义处理（比如添加字段、修改值），处理完后一定要把字典转回标准双引号的JSON字符串：

import json

def process(time, rdd):
    print("========= %s =========" % str(time))
    try:
        sqlContext = getSqlContextInstance(rdd.context)
        # 把字典转成标准JSON字符串
        json_str_rdd = rdd.map(lambda tweet_dict: json.dumps(tweet_dict))
        jsonRDD = sqlContext.read.json(json_str_rdd)
        jsonRDD.registerTempTable("tweets")
        jsonRDD.printSchema()
    except Exception as e:
        print(f"处理出错: {str(e)}")
        pass

rawKafkaStream = KafkaUtils.createStream(ssc, zkQuorum, "kafka-consumer", {kafkaTopic: 4})
# 先解析成字典做自定义处理
parsed_stream = rawKafkaStream.map(lambda rawTweet: json.loads(rawTweet[1]))
# 示例：添加自定义字段
# parsed_stream = parsed_stream.map(lambda d: {**d, "processing_time": str(time)})
parsed_stream.foreachRDD(process)

额外注意事项 #

• Spark 1.6的JSON Schema推断对复杂嵌套结构（比如Twitter的user、entities字段）支持有限，如果自动推断的Schema不符合预期，你可以手动用StructType和StructField定义Schema，再传给read.json(rdd, schema=your_custom_schema)。
• 要确保Kafka传来的每条消息都是完整的单条Twitter JSON，没有被拆分或包含换行符，否则也会导致解析失败。

内容的提问来源于stack exchange，提问作者analyticalpicasso