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

Posted 月影舞华

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了java及spark2.X连接mongodb3.X单机或集群的方法(带认证及不带认证)相关的知识,希望对你有一定的参考价值。

首先,我们明确的是访问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)

 

 









以上是关于java及spark2.X连接mongodb3.X单机或集群的方法(带认证及不带认证)的主要内容,如果未能解决你的问题,请参考以下文章

新闻实时分析系统 Spark2.X环境准备编译部署及运行

Spark2.x 与 Spark1.x 关系

Spark2.x+Python大数据机器学习视频课程

Spark2.x+Python大数据机器学习视频课程

spark2.x由浅入深深到底系列六之RDD java api详解二

spark2.x由浅入深深到底系列六之RDD java api详解四