spark/hadoop整合mongodb

MongoDB是一个文档型数据库，它可以方便的应用于大多数语言，其次是实现是C++，根据相关人员的测试证明mongodb的查询性能要好于现在市面上好多nosql数据库，相关的测试连接如下：

http://www.kuqin.com/shuoit/20140928/342398.html
下面简单介绍mongodb：
一、Mongodb特性
1、模式自由，支持动态查询、完全索引，可轻易查询文档中内嵌的对象及数组。
2、面向集合存储，易存储对象类型的数据，包括文档内包括文档内嵌对象及数组。
3、高效的数据存储，支持二进制数据及大型对象
4、支持复制和故障恢复：提供了主-从、主-主模式的数据复制及服务器之间的数据复制
5、自动分片以支持云级别的伸缩性，支持水平的数据库集群，可动态添加额外的服务器。
二、使用场景
1、适合作为信息基础设施的持久化缓存层
2、适合实时的插入，更新与查询，并具备应用程序实时数据存储所需的复制及高度伸缩性。
3、Mongodb的BSON数据格式非常适合文档化格式的存储及查询
4、适合由数十或数百台服务器组成的数据库。因为Mongodb已经包含了对MapReduce引擎的内置支持
三、不适合的场景
1、要求高度事务性的系统
2、传统的商业智能应用
3、复杂的跨文档（表）级联查询。
现在面临着大数据时代的挑战，下面开始讲述spark计算框架整合mongodb的使用。
先讲述spark读取mongodb：本人常用有两个方法，分别开始介绍：

spark整合mongodb之从mongodb读取：

 //方案1
val mongoConfig = new Configuration( )
mongoConfig.set("mongo.input.uri", "mongodb://master:20000,slave1:20000,slave2:20000/yang.relation2")
     mongoConfig.set( "mongo.input.split_size", "32" )//输入的大小
     mongoConfig.set( "mongo.input.split.read_shard_chunks", "true" )//读取分片
     mongoConfig.set( "mongo.input.fields{\"srcid\":\"1\",\"dstid\":\"1\"}" ) //读取的时候只读取自己需要的列 1表示读出，0表示不需要类似mongodb里面的projecttion
mongoConfig.set( "mongo.input.query","{\"dstid\":{\"$gt\":\"0\"}}" )      
val readfile = sc.newAPIHadoopRDD( mongoConfig, classOf[ MongoInputFormat ], classOf[ Object ],classOf[ BSONObject ] )
readfile.count( )


//方案2
 val sqlContex = new SQLContext( sc )
 val builder = MongodbConfigBuilder(Map(Host -> Host -> List("master:27017","slave1:27017","slave2:27017"), Database -> "graphdb",
  Collection -> "mongo", SamplingRatio -> 1.0, WriteConcern -> MongodbWriteConcern.Normal ) )
 val mconf = builder.build( )
 val readfile2 = sqlContex.fromMongoDB( mconf )
readfile2.count()

spark整合mongodb之写入mongodb：
方案1：

val mongoConfig = new Configuration()
    mongoConfig.set("mongo.auth.uri","mongodb://"+ userName +":"+ pwd+"@"+hosts+"/admin")
    mongoConfig.set("mongo.output.uri","mongodb://"+ hosts + "/GRAPHDB.DB_GRAPH")
    saveRdd.saveAsNewAPIHadoopFile("", classOf[Object], classOf[BSONObject],
      classOf[MongoOutputFormat[Object, BSONObject]], mongoConfig)


方案2：

import MongodbConfig._
import com.mongodb.casbah.{WriteConcern => MongodbWriteConcern, MongoClient}
import com.stratio.provider.mongodb._
 val sqlContext = new SQLContext( sc )
 val property = Array("id","name","age","sex","info")
 val dataFrame = sqlContext.createDataFrame(  data ).toDF( property:_*)
 val builder = MongodbConfigBuilder(Map(Host -> List("master:27017","slave1:27017","slave2:27017"), Database -> "test",
      Collection -> "test", SamplingRatio -> 1.0, WriteConcern -> MongodbWriteConcern.Normal))
 val mongoConf = builder.build()
 val dataFrame: DataFrame = sqlcontex.createDataFrame( rdd )
 dataFrame.saveToMongodb(mongoConf,true)
方案3：
利用rdd的foreachPartition在每个paritition建立连接，导入数据，此时如果分区输比较多，分配给spark的cpu核数比较多的话，会出现很多问题，比如：在查看mongodb日志的时候，mongos进程有时候会挂掉，是因为mongodb在分配读写锁的时候出现了问题，而且还会出现OOM（无法创建本地线程，这一点本小白正在解决）。一定要在里面创建连接哟，否则会出现序列化问题。

hadoo整合mongodb更新：
val mongoConfig = new Configuration()
mongoConfig.set(“mongo.output.uri”,”mongodb://master:27017/db.table”)
saveRdd.saveAsNewAPIHadoopFile(“”, classOf[ Object ], classOf[ MongoUpdateWritable ],
classOf[ MongoOutputFormat[ Object,MongoUpdateWritable ] ],mongoConfig ).

更新的时候可以结合mongodb的数值修改器使用。以后有时间了给大家分享数据修改器的使用。本人小白一枚，如果有问题，希望大家给予指出，小杨在这里拜谢各位大神了。