spark2.x向mongodb中讀取寫入資料，讀取寫入相關引數參考https://docs.mongodb.com/spark-connector/current/configuration/#cache-configuration

從mongodb中讀取資料時指定資料分割槽欄位，分割槽大小提高讀取效率， 當需要過濾部分資料集的情況下使用Dataset/SQL的方式filter，Mongo Connector會建立aggregation pipeline在mongodb端進行過濾，然後再傳回給spark進行優化處理

val spark = SparkSession.builder
          .appName(this.getClass.getName().stripSuffix("$"))
          .getOrCreate()
val inputUri="mongodb://test:
 
[email protected]:27017/test.articles"
val df = spark.read.format("com.mongodb.spark.sql").options(
             Map("spark.mongodb.input.uri" -> inputUri,
                 "spark.mongodb.input.partitioner" -> "MongoPaginateBySizePartitioner",
                 "spark.mongodb.input.partitionerOptions.partitionKey"  -> "_id",
                 "spark.mongodb.input.partitionerOptions.partitionSizeMB"-> "32"))
        .load()
val currentTimestamp = System.currentTimeMillis()
val originDf = df.filter(df("updateTime") < currentTimestamp && df("updateTime") >= currentTimestamp - 1440 * 60 * 1000)
                     .select("_id", "content", "imgTotalCount").toDF("id", "content", "imgnum")
 

val outputUri="mongodb://test:[email protected]:27017/test.article_garbage" 
saveDF.write.options(Map("spark.mongodb.output.uri"-> outputUri))
          .mode("append")
          .format("com.mongodb.spark.sql")
          .save()
 

spark操作mongodb的scala-api文件：https://docs.mongodb.com/spark-connector/current/scala-api/