本文不支持复制粘贴的转载，鼓励修改和扩展后的转载。转载前必须邮件取得本人同意，联系方式：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()
    }
  }
}

还没完成。