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

Posted 2020-10-04 月影舞华

首先，我们明确的是访问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:[email protected]:27017/?authSource=admin")
    //val mongoURI = new MongoClientURI("mongodb://192.168.2.48:27017/");
    lazy val mongo = new MongoClient(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:[email protected]: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:[email protected]:27017/?authSource=test",
              "database" -> "test",
              "collection" -> "test"
              ))

val rdd = MongoSpark.builder().sparkSession(spark).readConfig(readConfig).build().toRDD()
//val r2 = MongoSpark.load(spark.sparkContext, readConfig)