Spark连接MongoDB之Scala

小风儿

import org.apache.spark.sql.SparkSession　　
import com.mongodb.spark._
　　
import com.mongodb.spark.config._
　　
import org.bson.Document
　　

　　val spark = SparkSession.builder()
　　.master("local")
　　.appName("MongoSparkConnector")
　　.config("spark.some.config.option", "some-value")
　　.getOrCreate()
　　

　　
val uri = "mongodb://172.1.1.1:27017"
　　

　　val userDF = spark.sql("""
　　
select
　　uid,
　　name,
　　current_date() version
　　from test_table
　　limit 100
　　""").repartition(8)
　　

　　
// Write to MongoDB
　　userDF.write.mode("overwrite").format("com.mongodb.spark.sql").options(
　　Map(
　　"uri" -> uri,
　　"database" -> "test",
　　"collection" -> "test_table")).save()
　　

　　
// Read From MongoDB
　　val df = spark.read.format("com.mongodb.spark.sql").options(
　　Map(
　　"uri" -> uri,
　　"database" -> "test",
　　"collection" -> "test_table")).load()
　　

　　
// 其他方式
　　userDF.write.mode("overwrite").format("com.mongodb.spark.sql").options(
　　Map(
　　"spark.mongodb.input.uri" -> uri,
　　"spark.mongodb.output.uri" -> uri,
　　"spark.mongodb.output.database" -> "test",
　　"spark.mongodb.output.collection" -> "test_table")).save()
　　

　　MongoSpark.save(
　　userDF.write.mode("overwrite").options(
　　Map(
　　"spark.mongodb.input.uri" -> uri,
　　"spark.mongodb.output.uri" -> uri,
　　"spark.mongodb.output.database" -> "test",
　　"spark.mongodb.output.collection" -> "test_table")))
　　

　　MongoSpark.save(
　　userDF.write.mode("overwrite").options(
　　Map(
　　"uri" -> uri,
　　"database" -> "test",
　　"collection" -> "test_table")))
　　

　　spark.stop()