spark连接MongoDB的两种方式

拼搏现实的明天。 2023-11-11 06:57 175阅读 0赞

#### 一、准备机器 ####

操作系统：centos 6  
机器：10.211.55.7  
端口：27017  
二、安装  
1、下载MongoDB（64位）  
wget http://fastdl.mongodb.org/linux/mongodb-linux-x86\_64-2.4.9.tgz  
2、安装MongoDB（安装到/data/program/software）  
tar –zxf mongodb-linux-x86\_64-2.4.9.tgz  
mv mongodb-linux-x86\_64-2.4.9 mongodb  
cd mongodb  
mkdir db  
mkdir logs  
cd bin

--------------------

vi mongodb.conf  
写入如下：  
dbpath=/data/program/software/mongodb/db  
logpath=/data/program/software/mongodb/logs/mongodb.log  
port=27017  
fork=true  
nohttpinterface=true  
设置参数解释：  
dbpath:mongod进程存储数据目录，此配置仅对mongod进程有效。默认值为：/data/db  
logpath:日志目录  
port:mongod侦听端口，默认值为27017  
fork:后台方式运行mongodb  
nohttpinterface:关闭http接口

--------------------

3、重新绑定mongodb的配置文件地址和IP  
/data/program/software/mongodb/bin/mongod --bind\_ip 10.211.55.7 -f  
/data/program/software/mongodb/bin/mongodb.conf  
4、加入开机启动mongodb  
vi /etc/rc.d/rc.local  
/data/program/software/mongodb/bin/mongod --bind\_ip 10.211.55.7 -f  
/data/program/software/mongodb/bin/mongodb.conf  
5、进入mongodb看看是否能连接  
/data/program/software/mongodb/bin/mongo 10.211.55.7  
\#查看数据列表  
show dbs  
\#查看当前db版本

#### Spark连接Mongodb代码实现 ####

有两种实现方式：

##### 第一种：用如下这个jar，将其加入到pom文件中 #####

org.mongodb.spark mongo-spark-connector\_2.11 2.0.0 导入包 import com.mongodb.spark.\_; 基本配置讲解 val conf=new SparkConf().setAppName("mongodbTest") .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //定义序列化方式 .set("spark.mongodb.input.uri", "mongodb://10.211.55.7:27017/bigdata.person") //定义输入地址 .set("spark.mongodb.output.uri", "mongodb://10.211.55.7:27017/bigdata.person") //定义输出地址 .set("spark.mongodb.keep\_alive\_ms", "10000000") //定义链接时间 由于之前安装的mongodb版本低于2.6，所以需要从新安装一下mongodb，将版本升级

mongodb地址：https://fastdl.mongodb.org/linux/mongodb-linux-x86\_64-3.4.7.tgz

使用地址介绍：https://docs.mongodb.com/spark-connector/master/configuration/

##### 第二种方式导入如下jar，加入到pom文件中 #####

com.stratio.datasource spark-mongodb\_2.11 0.12.0 连接mongodb配置如下 @transient private var instance:MongodbConfigBuilder=\_; def getInstance():MongodbConfigBuilder=\{ if(instance==null)\{ instance=MongodbConfigBuilder( Map(Host -> List("10.211.55.7:9927"), Database -> "bigdataml", Collection -> "product\_type\_14000\_16000", SamplingRatio -> 1.0, WriteConcern ->"normal", ConnectionsTime->"180000", ConnectTimeout->"50000" )) \} instance \}

调用  
val builder = MongodbSingleton.getInstance() result.saveToMongodb(builder.build());  
对比spark和mongodb版本进行选择

插件地址：https://github.com/Stratio/spark-mongodb