【Spark五十】Spark读写MongoDB

 

 

1. Spark写入MongoDB的实例代码

如下代码实现将RDD写入到MongoDB的spark数据库的oc集合(Collection)中

 

 

package spark.examples.db

import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkContext, SparkConf}
import com.mongodb.hadoop.MongoOutputFormat
import org.apache.spark.SparkContext._
import org.bson.BasicBSONObject

object SparkMongoDBIntegration {

  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("SparkRDDCount").setMaster("local")
    val sc = new SparkContext(conf);
    val data = sc.parallelize(List(("Tom", 31), ("Jack", 22), ("Mary", 25)))
    val config = new Configuration()
    //spark.oc指的是，spark是MongoDB的数据库的名字，而ic表示数据库中的一个Collection
    config.set("mongo.input.uri", "mongodb://192.168.26.137:27017/spark.ic")
    config.set("mongo.output.uri", "mongodb://192.168.26.137:27017/spark.oc")
    //使用MongoOutputFormat将数据写入到MongoDB中
    val rdd = data.map((elem) => {
      val obj = new BasicBSONObject()
      obj.put("name", elem._1)
      obj.put("age", elem._2)
      //转换后的结果，KV对
      //第一个是BSON的ObjectId，插入时可以指定为null，MongoDB Driver在插入到MongoDB时，自动的生成
      //obj是BSONObject，是MongoDB Driver接收的插入对象
      (null, obj)
    })
    //RDD data is a KV pair,so it can use saveAsNewAPIHadoopFile
    rdd.saveAsNewAPIHadoopFile("file:///bogus", classOf[Any], classOf[Any], classOf[com.mongodb.hadoop.MongoOutputFormat[Any, Any]], config)
  }
}

 

例子程序使用了MongoDB提供的Hadoop-MongoDB-Connector来实现将RDD写入到MongoDB中，注意的是，运行这个程序无需启动Hadoop（指的是HDFS），也就是把MongoDB作为一个外部数据源，通调用rdd.saveAsTextFile("file:///localfilesystemfile")的道理一样，通过这个Connector，MongoDB实现了作为HDFS格式兼容的数据源，因此Spark可以直接操作之。

 

2. 代码的依赖

上面的代码要运行，依赖两个jar包

• MongodB-Hadoop-Connector
• MongoDB Driver

分别可以从如下地址下载：https://github.com/plaa/mongo-spark和https://github.com/mongodb/mongo-java-driver/releases

 

3.关于mongo.output.uri和mongo.input.uri

它们的含义可以参见http://www.itpub.net/thread-1808762-1-1.html

启动mongo.output.uri的意思是指specify an output collection URI to write the output to. If using a sharded cluster, you can specify a space-delimited list of URIs referring to separate mongos instances and the output records will be written to them in round-robin fashion to improve throughput and avoid overloading a single mongos.

也就是说，mongo.output.uri是一个Spark输出数据的集合的uri，例子中"mongodb://192.168.26.137:27017/spark.ic"，spark.ic的含义是spark是MongoDB的数据库的名字，而ic表示数据库中的一个Collection

 

4. 数据库和集合需要预先建立吗？

在上面的例子中，spark数据库和ic/oc集合都不需要预先建立，MongoDB Driver会首先判断，要写入数据的数据库和集合是否存在，如果不存在则会首先建立

 

5. 如何设置数据库的用户名密码？

在实际工作中，MongoDB的数据库都是有用户名密码的，也就是在指定mongo.output.uri和mongo.input.uri时，都要带着用户名和密码信息，这如何指定？

可以使用如下方式来指定：

 

 

mongodb://joe:123456@192.168.26.137:27017/spark.ic?readPreference=secondary

使用joe和123456作为用户名密码认证，同时指定读优先策略为secondary(从slave节点上读取）

 

 

参考：

http://codeforhire.com/2014/02/18/using-spark-with-mongodb/

http://my.oschina.net/1987times/blog/265483

https://github.com/plaa/mongo-spark