python之pyspark 将 twitter json 流式传输到 DF

我正在使用 python API 将 spark-streaming 与 twitter 集成。我看到的大多数示例或代码片段和博客是，他们从 Twitter JSON 文件中提取几列进行最终操作。但根据我的用例，我需要 twitter JSON 的所有字段并将其转换为数据帧。这就是我面临的问题，因为我 sqlContext.read.json() 正在将整个 JSON DStream 转储到 _corrupt_record

+--------------------+ 
|     _corrupt_record| 
+--------------------+ 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 
|{u'quote_count': ...| 

此外，这个问题似乎可以通过使用 spark 2+ 版本的 structured streaming 来解决。但我必须坚持使用 spark 1.6。以下是我的代码片段。

def process(time, rdd): 
    print("========= %s =========" % str(time)) 
    try: 
 
        sqlContext = getSqlContextInstance(rdd.context) 
 
        jsonRDD = sqlContext.read.json(rdd) 
        jsonRDD.registerTempTable("tweets") 
        jsonRDD.printSchema() 
    except: 
        pass 
 
 
rawKafkaStream = KafkaUtils.createStream(ssc, zkQuorum, "kafka-consumer", {kafkaTopic: 4}) 
parsed_stream = rawKafkaStream.map(lambda rawTweet: json.loads(rawTweet[1])) 
 
parsed_stream.foreachRDD(process) 

请您参考如下方法:

Python json.dumps() 在 Spark 中创建 RDD[Dict] 类型的字典 RDD。为了使其成为 DF，以下行将起作用

SQLContext.jsonRDD(RDD[dict].map(lambda x: json.dumps(x))) 

为了让它在我的案例中起作用，我必须执行以下操作

def process(time, rdd): 
    print("========= %s =========" % str(time)) 
    try: 
 
        sqlContext = getSqlContextInstance(rdd.context) 
 
        jsonRDD=sqlContext.jsonRDD(rdd.map(lambda x: json.dumps(x))) 
        jsonRDD.registerTempTable("tweets") 
        jsonRDD.printSchema() 
    except: 
        pass 
 
 
rawKafkaStream = KafkaUtils.createStream(ssc, zkQuorum, "kafka-consumer", {kafkaTopic: 4}) 
parsed_stream = rawKafkaStream.map(lambda rawTweet: json.loads(rawTweet[1])) 
 
parsed_stream.foreachRDD(process) 

有关此方法的更多详细信息。引用link