【发布时间】：2018-08-02 11:54:36
【问题描述】：
我正在使用 PySpark 并从 Kafka Broker 获取数据。
下面的代码可以帮助我做到这一点：
import json
import sys
from pyspark import Row
from pyspark.sql.types import *
from pyspark import SparkContext
from pyspark.sql.functions import *
from pyspark.conf import SparkConf
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, TopicAndPartition
from pymongo import MongoClient
# Spark Context created here
sc = SparkContext("local[*]", appName="TwitterStreamKafka")
#Spark Streaming Context Created Here
ssc = StreamingContext(sc, 1)
#Kafka Stream Created Here
tweets = KafkaUtils.createStream(ssc, "localhost:2181", "spark-streaming-consumer", {"twitter": 1})
tweets.pprint()
ssc.start()
ssc.awaitTermination()
使用以下代码从推文中提取 JSON：
tweet_json = tweets.map(lambda x: json.loads(x[1]))
现在我想将这个 tweet_json 插入 MongoDB。但不知道该怎么做。
检查了 Spark-MongoConnector，它说它需要 DataFrame 存储在 MongoDB 中
但是 tweet_json 的类型是 "pyspark.streaming.dstream.TransformedDStream"
如何将其转换为 Dataframe 以存储到 MongoDB？
或
如何使用 PySpark 将获取的推文保存到 MongoDB
提前谢谢你！
在程序中建议编辑后，我按以下方式对其进行了编辑：
import json
import sys
from pyspark import Row
from pyspark.sql.types import *
from pyspark import SparkContext
from pyspark.sql.functions import *
from pyspark.conf import SparkConf
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, TopicAndPartition
from pymongo import MongoClient
def convert(rdd):
df_json = rdd.map(lambda x: json.loads(x[1])).toDF()
return df_json
# Spark Context created here
sc = SparkContext("local[*]", appName="TwitterStreamKafka")
#Spark Streaming Context Created Here
ssc = StreamingContext(sc, 1)
#Kafka Stream Created Here
tweets = KafkaUtils.createStream(ssc, "localhost:2181", "spark-streaming-consumer", {"twitter": 1})
#Tweets Printed Here
tweets.pprint()
tweets.foreachRDD(lambda rdd: convert(rdd).write\
.format('com.mongodb.spark.sql.DefaultSource').mode('append')
.option('database',Twitter).option('collection',RestTwitter).save())
ssc.start()
ssc.awaitTermination()
然后按以下方式初始化spark提交：
spark-submit --conf "spark.mongodb.inuri=mongodb://127.0.0.1/TwitterDB.RestDatareadPreference=primaryPreferred" --conf "spark.mongodb.output.uri=mongodb://127.0.0.1/TwitterDB.RestData" --packages org.mongodb.spark:mongo-spark-connector_2.11:2.0.0 --jars spark-streaming-kafka-0-8-assembly_2.11-2.0.0.jar PySparkTwitter.py
它引发了以下错误：
文件“/home/hduser/test/PySparkTwitter.py”，第 24 行，转换中 df_json = rdd.map(lambda x: json.loads(x[1])).toDF() AttributeError：“PipelinedRDD”对象没有属性“toDF”
CreateStream的数据格式：
(None, '{"lang": "en", "id": 967004613332303873, "favorited": false, "possibly_sensitive": false, "is_quote_status": false, "geo": null, "user": {"lang": "en", "profile_use_background_image": false, "profile_banner_url": "https://pbs.twimg.com/profile_banners/19908749/1517325981", "is_translator": false, "id": 19908749, "profile_sidebar_border_color": "FFFFFF", "favourites_count": 1912, "profile_background_tile": false, "profile_image_url_https": "https://pbs.twimg.com/profile_images/921307498409230336/VWkCvBcu_normal.jpg", "friends_count": 1960, "profile_background_image_url": "http://pbs.twimg.com/profile_background_images/222053103/twitter-image-bg.jpg", "has_extended_profile": false, "profile_link_color": "AB0D0D", "screen_name": "ZNConsulting", "geo_enabled": true, "url": "", "profile_text_color": "000000", "default_profile": false, "utc_offset": 3600, "is_translation_enabled": false, "statuses_count": 5833, "profile_background_image_url_https": "https://pbs.twimg.com/profile_background_images/222053103/twitter-image-bg.jpg", "verified": false, "name": "ZN Consulting", "notifications": false, "protected": false, "id_str": "19908749", "translator_type": "none", "profile_image_url": "http://pbs.twimg.com/profile_images/921307498409230336/VWkCvBcu_normal.jpg", "created_at": "Mon Feb 02 14:28:00 +0000 2009", "time_zone": "Brussels", "listed_count": 253, "follow_request_sent": false, "location": "Brussels; Our website:", "profile_background_color": "F7F7F7", "contributors_enabled": false, "entities": {"url": {"urls": [{"display_url": "znconsulting.com", "expanded_url": "http://znconsulting.com", "url": "", "indices": [0, 23]}]}, "description": {"urls": []}}, "default_profile_image": false, "following": false, "followers_count": 2312, "description": "The #digital communication agency in #Brussels. Strategy, digital campaigns, analysis & #socialmedia. #Hyperthinking to give you the #DigitalAdvantage \\ud83d\\ude0e", "profile_sidebar_fill_color": "C7C7C7"}, "in_reply_to_user_id_str": null, "contributors": null, "retweet_count": 0, "text": "Following user feedback, #Google is now blocking intrusive ads with a built-in adblocker in #Chrome \\u26d4\\n\\n", "retweeted": false, "truncated": false, "in_reply_to_user_id": null, "id_str": "967004613332303873", "source": "<a href=\\"https://about.twitter.com/products/tweetdeck\\" rel=\\"nofollow\\">TweetDeck</a>", "created_at": "Fri Feb 23 11:54:00 +0000 2018", "metadata": {"iso_language_code": "en", "result_type": "recent"}, "in_reply_to_screen_name": null, "in_reply_to_status_id_str": null, "entities": {"symbols": [], "urls": [{"display_url": "theguardian.com/technology/201\\u2026", "expanded_url": "https://www.theguardian.com/technology/2018/feb/15/google-adblocker-chrome-browser", "url": "", "indices": [103, 126]}], "user_mentions": [], "hashtags": [{"text": "Google", "indices": [25, 32]}, {"text": "Chrome", "indices": [92, 99]}]}, "coordinates": null, "in_reply_to_status_id": null, "place": null, "favorite_count": 0}')
【问题讨论】：
标签： json mongodb apache-spark pyspark spark-dataframe