本文不支持复制粘贴的转载，鼓励修改和扩展后的转载。转载前必须邮件取得本人同意，联系方式：1505514388@qq.com

各种语言和框架连接 mongodb 的基础代码#

使用 Java 客户端连接 mongodb#

        // 获取 mongo 客户端
        MongoClient mongoClient=new MongoClient("localhost");
        // 获取数据库
        MongoDatabase database=mongoClient.getDatabase("test_database");
        // 获取集合
        MongoCollection<Document>collection=database.getCollection("test_database");
        // 查询集合中的第一个元素
        Document myDoc=collection.find().first();
        System.out.println(myDoc);
        // 关闭资源
        mongoClient.close();

使用 scala 客户端连接 mongodb#

    val mongoClient = MongoClient("mongodb://localhost:27017")
    val database = mongoClient.getDatabase("test_basedata")
    val collection = database.getCollection("test_basedata")
    val document = collection.find().first()
    System.out.println(document.toString)
    mongoClient.close()

使用 spark 连接 mongodb#

在单纯的使用 scala 的 mongodb 的连接器时遇到了以下报错：

com.mongodb.ConnectionString.getThreadsAllowedToBlockForConnectionMultiplier()Ljava/lang/Integer;

解决办法为：

<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>mongo-java-driver</artifactId>
    <version>3.10.0</version>
</dependency>

具体原因未知
spark 的连接器的代码如下：

    //TODO 开启环境
    val spark = SparkSession.builder().master("local")
      .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/test_database.test_database")
      .config("spark.mongodb.output.uri", "mongodb://127.0.0.1/test_database.test_database")
      .getOrCreate()
   //TODO 数据操作
    val testDF = MongoSpark.load(spark)
    testDF.show(20)
    //TODO 关闭环境
    spark.close()

使用 flink 将 mongodb 作为数据源#

在 flink 中没有将 mongodb 作为数据源的，所以下面使用的依赖也是第三方连接器。
在一般情况下，也不会遇到将 mongodb 作为 flink 的数据源。
所需要的依赖：

<!-- https://mvnrepository.com/artifact/org.mongodb/casbah-core -->
<dependency>
    <groupId>org.mongodb</groupId>
    <artifactId>casbah-core_2.11</artifactId>
    <version>3.1.1</version>
</dependency>

下面是自定义的 source

package com.myFlink.test
import com.mongodb.{BasicDBObject, MongoClientURI, casbah}
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
class MongodbSource extends RichSourceFunction[User]{
// 创建 mongodb 数据源时运行
var client:MongoClient=_
override def open(parameters: Configuration): Unit = {
// 连接本地 mongodb
client= MongoClient(casbah.MongoClientURI("mongodb://localhost:27017"))
}
// 实时运行的函数
override def run(sourceContext: SourceFunction.SourceContext[User]): Unit = {
//TODO 获取数据库
val database = client("test_database")
//TODO 获取集合
val coll = database("test_database")
//TODO 取出数据
val query = new BasicDBObject("date", "20171203")
val cursorType = coll.find(query)
if(cursorType.nonEmpty){
val oneData = cursorType.next() // 拿出一条数据
sourceContext.collect(
User(
name=oneData.get("name").toString,
date=oneData.get("date").toString
)
)
}
}
// 结束时的函数
override def cancel(): Unit ={
if(client!=null){
client.close()
}
}
}
case class User(name:String,date:String)

然后向本地的 mongdb 中插入数据：

db.getCollection("test_database").insert({"name":"kone", "date":"20171203"})

然后输出：

4> User(kone,20171203)

使用 flink 将 mongodb 作为 sink#

package com.myFlink.test
import com.mongodb.BasicDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
class MongoDBSink extends RichSinkFunction[User]{
var client:MongoClient=_
override def open(parameters: Configuration): Unit = {
client=MongoClient(MongoClientURI("mongodb://localhost:27017"))
}
override def invoke(value: User): Unit = {
//TODO 获取数据库
val database = client("test_database")
//TODO 获取集合
val coll = database("test_database")
//TODO 数据操作
val obj = new BasicDBObject("name", value.name).append("date", value.date)
coll.insert(obj)
}
override def close(): Unit = {
if(client!=null){
client.close()
}
}
}

还没完成。