标签：roc 数据结构 min 3.x 方式 迭代器 document blog mat
首先，我们明确的是访问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 = 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: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)
java及spark2.X连接mongodb3.X单机或集群的方法（带认证及不带认证）
标签：roc 数据结构 min 3.x 方式 迭代器 document blog mat
原文地址：http://www.cnblogs.com/gaoze/p/7419458.html
