how to save dataframe in mongodb using pyspark?

Question

my_spark = SparkSession \ .builder \ .appName("DGLE") \ 
 config("spark.mongodb.input.uri", "mongodb://127.0.0.1/local.DGLE") \ getOrCreate() 
JDBC.write.format("com.mongodb.spark.sql.DefaultSource").option("spark.mongodb.output.uri", "mongodb://127.0.0.1:27017/local.DGLE").save()

This is my code, do not know where and what is the error, please help

my_spark = SparkSession \ builder \ .appName("DGLE") \ .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/local.DGLE") \ .getOrCreate() 
JDBC.write.format("com.mongodb.spark.sql.DefaultSource").option("spark.mongodb.output.uri", "mongodb://127.0.0.1:27017/local.DGLE").save()

Could you please explain you problem more. not just the code — El.Hum
– El.Hum, Commented Jan 12, 2019 at 8:01
Also please your code. I tried to edit but I dont really know how it should exactly be. Please edit your question unless I have to flag it — El.Hum
– El.Hum, Commented Jan 12, 2019 at 8:04

Kriti Pawar · Accepted Answer · 2020-05-18 15:50:25Z

Efficient way to write into mongodb from pyspark is to use MongoDB Spark Connector. Connector will convert the data into BSON format and save it to mongodb. Let's say you have spark dataframe named df which you want to save in mongodb. You can try:

from pyspark.sql import SparkSession, SQLContext
from pyspark import SparkConf, SparkContext
sc = SparkContext()
spark = SparkSession(sc)   


df.write.format("com.mongodb.spark.sql.DefaultSource").mode("append").option("spark.mongodb.output.uri","mongodb://username:password@server_details:27017/db_name.collection_name?authSource=admin").save()

If you are using notebook write this at the top-

%%configure
{"conf": {"spark.jars.packages": "org.mongodb.spark:mongo-spark-connector_2.11:2.3.2"}}

If you are using spark-submit command:

spark-submit --conf spark.pyspark.python=/usr/bin/anaconda/envs/py35/bin/python3.5 --packages org.mongodb.spark:mongo-spark-connector_2.11:2.3.1 file_name.py

Pardeep · Accepted Answer · 2021-04-16 05:01:13Z

0

def save(message: DataFrame):
    message.write \
        .format("mongo") \
        .mode("append") \
        .option("database", "db_name") \
        .option("collection", "collection_name") \
        .save()
    pass

And this is how the SparkSession will look like

spark: SparkSession = SparkSession \
    .builder \
    .appName("MyApp") \
    .config("spark.mongodb.input.uri", "mongodb://localhost:27017") \
    .config("spark.mongodb.output.uri", "mongodb://localhost:27017") \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:3.0.1") \
    .master("local") \
    .getOrCreate()

edited Apr 16, 2021 at 5:01

answered Apr 15, 2021 at 11:28

Pardeep

1,00511 silver badges18 bronze badges

Collectives™ on Stack Overflow

how to save dataframe in mongodb using pyspark?

2 Answers 2

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related