首先，我们明确的是访问Mongos和访问单机Mongod并没有什么区别。接下来的方法都是既可以访问mongod又可以访问Mongos的。

另外，读作java写作scala，反正大家都看得懂......大概？

1、不带认证集群的连接方法(JAVAscala)：

首先是创建连接的方法，我们先声明一个client，然后指定访问的DB和collection：

private lazy val mongo = new MongoClient("192.168.2.51", 27017)private lazy val db = mongo.getDatabase("test")private lazy val dbColl = db.getCollection("origin2")

然后我们读取数据：

import com.mongodb.client.model.Filters.{eq =>eqq}

val docs= dbColl.find(eqq("basiclabel.procedure", "second")).iterator()

额。。上面那段代码是带filter过滤的读取数据。首先Import com.mongodb.client.model.Filters.eq并把eq重命名为eqq，然后通过dbColl.find(Bson)方法读取指定数据。剩下的就是正常的迭代器的使用方法了，docs获取出来的数据是Iterator[Document]。

然后我们更新数据：

dbColl.updateOne(eqq("_id", x.get("_id")), set("segdata", fenduan(str, name)))

上面这段代码是说找到_id对应的数据，并将其中一个字段set为一个新的值，这个值可以为Document，String，Int，List等一系列数据结构。我这里fenduan方法返回的是一个Document，做了一层嵌套。

至于插入数据更为简单：

dbColl.insertOne(doc)

2、不带认证的spark读取方法(scala，理直气壮)

两种方式，其一是在创建sparksession的时候(SparkContext可以使用第二种方法，醒醒兄弟，2017年了)，直接指定"spark.mongodb.input.uri"。然后使用正常的MongoSpark来读取数据。(pipeline里面是过滤条件，愿意尝试的各位可以自己试试filter下的其他方法)。使用rdd是因为rdd更适合进行map和flatmap等一系列精细的转换操作，如果只需要读数据，可以使用MongoSpark.read(spark)方法，直接获取DataFrameReader。

val spark =SparkSession.builder()

.master("spark://192.168.2.51:7077")

.config(new SparkConf().setJars(Array("hdfs://192.168.2.51:9000/mongolib/mongo-spark-connector_2.11-2.0.0.jar","hdfs://192.168.2.51:9000/mongolib/bson-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/mongo-java-driver-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/mongodb-driver-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/mongodb-driver-core-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/commons-io-2.5.jar","hdfs://192.168.2.51:9000/segwithorigin2.jar")))

.config("spark.cores.max", 80)

.config("spark.executor.cores", 16)

.config("spark.executor.memory", "32g")

.config("spark.mongodb.input.uri", "mongodb://192.168.2.51:27017/test.origin2")//.config("spark.mongodb.output.uri", "mongodb://192.168.12.161:27017/test.origin2")

.getOrCreate()

val rdd = MongoSpark.builder().sparkSession(spark).pipeline(Seq(`match`(eqq("basiclabel.procedure", "second")))).build.toRDD()

第二种方式也较为简单，创建一个ReadConfig，这个是connector提供的一个单例类，可以设置很多参数，例如(此时不必指定"spark.mongodb.input.uri")，如下所示是通过sparkcontext和通过sparksession两种方式读取数据的方法：

val readConfig =ReadConfig(Map("uri" -> "mongodb://192.168.2.48:27017/","database" -> "test","collection" -> "test"))

val r2=MongoSpark.load(spark, readConfig).rdd//val r2 = MongoSpark.load(spark.sparkContext, readConfig)

3、带认证的Java读取方法：

带认证的需要先创建一个MongoURI，在URI里把用户名，密码和认证库都指定清楚。这种方法通用性比较强，因为spark也这么用，如果使用其他方式认证要么是必须使用库等于认证库，要么是没有通用性。这种方法可以在admin认证然后去读test的数据，就很好。

//带认证的需要先创建一个MongoURI，在URI里把用户名，密码和认证库都指定清楚，至于为什么需要指定库建议看上一篇博客

val mongoURI = new MongoClientURI("mongodb://gaoze:gaolaoban@192.168.2.48:27017/?authSource=admin")//val mongoURI = new MongoClientURI("mongodb://192.168.2.48:27017/");

lazy val mongo = newMongoClient(mongoURI)private lazy val db = mongo.getDatabase("test")private lazy val dbColl = db.getCollection("test")

//然后和1一样

4、带认证的Spark读取方法：

同3一样，在URI里加入用户名密码和库就行了：

val spark =SparkSession.builder()

.master("spark://192.168.2.51:7077")

.config(new SparkConf().setJars(Array("hdfs://192.168.2.51:9000/mongolib/mongo-spark-connector_2.11-2.0.0.jar","hdfs://192.168.2.51:9000/mongolib/bson-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/mongo-java-driver-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/mongodb-driver-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/mongodb-driver-core-3.4.2.jar","hdfs://192.168.2.51:9000/mongolib/commons-io-2.5.jar","hdfs://192.168.2.51:9000/segwithorigin2.jar")))

.config("spark.cores.max", 80)

.config("spark.executor.cores", 16)

.config("spark.executor.memory", "32g")

//这里这个配置项指定了用户名gaoze，密码gaolaoban，认证库admin

.config("spark.mongodb.input.uri", "mongodb://gaoze:gaolaoban@192.168.2.51:27017/test.origin2?authSource=admin").getOrCreate()

val rdd= MongoSpark.builder().sparkSession(spark).pipeline(Seq(`match`(eqq("basiclabel.procedure", "second")))).build.toRDD()

或者：

//这里指定了用户名rw，密码1，认证库test

val readConfig =ReadConfig(Map("uri" -> "mongodb://rw:1@192.168.2.48:27017/?authSource=test","database" -> "test","collection" -> "test"))

val rdd = MongoSpark.builder().sparkSession(spark).readConfig(readConfig).build().toRDD()

//val r2 = MongoSpark.load(spark.sparkContext, readConfig)