java及spark2.X连接mongodb3.X单机或集群的方法（带认证及不带认证）

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

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

1、不带认证集群的连接方法（JAVAscala）：

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

privatelazyvalmongo=newMongoClient("192.168.2.51",27017)
privatelazyvaldb=mongo.getDatabase("test")
privatelazyvaldbColl=db.getCollection("origin2")

　然后我们读取数据：

importcom.mongodb.client.model.Filters.{eq=>eqq}
valdocs=dbColl.find(eqq("basiclabel.procedure","second")).iterator()

　　额。。上面那段代码是带filter过滤的读取数据。首先Importcom.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。

valspark=SparkSession.builder()
.master("spark://192.168.2.51:7077")
.config(newSparkConf().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()
valrdd=MongoSpark.builder().sparkSession(spark).pipeline(Seq(`match`(eqq("basiclabel.procedure","second")))).build.toRDD()

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

valreadConfig=ReadConfig(Map(
"uri"->"mongodb://192.168.2.48:27017/",
"database"->"test",
"collection"->"test"
))
　　　　valr2=MongoSpark.load(spark,readConfig).rdd
//　　　　valr2=MongoSpark.load(spark.sparkContext,readConfig)

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

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

//带认证的需要先创建一个MongoURI，在URI里把用户名，密码和认证库都指定清楚，至于为什么需要指定库建议看上一篇博客
valmongoURI=newMongoClientURI("mongodb://gaoze:gaolaoban@192.168.2.48:27017/?authSource=admin")
//valmongoURI=newMongoClientURI("mongodb://192.168.2.48:27017/");
lazyvalmongo=newMongoClient(mongoURI)
privatelazyvaldb=mongo.getDatabase("test")
privatelazyvaldbColl=db.getCollection("test")
//然后和1一样

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

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

valspark=SparkSession.builder()
.master("spark://192.168.2.51:7077")
.config(newSparkConf().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()
valrdd=MongoSpark.builder().sparkSession(spark).pipeline(Seq(`match`(eqq("basiclabel.procedure","second")))).build.toRDD()

或者：

//这里指定了用户名rw，密码1，认证库test
valreadConfig=ReadConfig(Map(
"uri"->"mongodb://rw:1@192.168.2.48:27017/?authSource=test",
"database"->"test",
"collection"->"test"
))

valrdd=MongoSpark.builder().sparkSession(spark).readConfig(readConfig).build().toRDD()
//valr2=MongoSpark.load(spark.sparkContext,readConfig)