Spark整合Mongodb的方法

spark介绍

按照官方的定义，spark 是一个通用，快速，适用于大规模数据的处理引擎。

通用性：我们可以使用spark sql来执行常规分析， spark streaming 来流数据处理， 以及用mlib来执行机器学习等。java，python，scala及r语言的支持也是其通用性的表现之一。

快速： 这个可能是spark成功的最初原因之一，主要归功于其基于内存的运算方式。当需要处理的数据需要反复迭代时，spark可以直接在内存中暂存数据，而无需像map reduce一样需要把数据写回磁盘。官方的数据表明：它可以比传统的map reduce快上100倍。

大规模：原生支持hdfs，并且其计算节点支持弹性扩展，利用大量廉价计算资源并发的特点来支持大规模数据处理。

环境准备

mongodb下载

解压安装

启动mongodb服务

$mongodb_home/bin/mongod --fork --dbpath=/root/data/mongodb/ --logpath=/root/data/log/mongodb/mongodb.log

pom依赖

<dependency> 
<groupid>org.mongodb.spark</groupid> 
<artifactid>mongo-spark-connector_2.11</artifactid> 
<version>${spark.version}</version> 
</dependency>

实例代码

object connapptest { 
def main(args: array[string]): unit = { 
val spark = sparksession.builder() 
.master("local[2]") 
.appname("connapptest") 
.config("spark.mongodb.input.uri", "mongodb://192.168.31.136/testdb.testcollection") // 指定mongodb输入 
.config("spark.mongodb.output.uri", "mongodb://192.168.31.136/testdb.testcollection") // 指定mongodb输出 
.getorcreate() 
// 生成测试数据 
val documents = spark.sparkcontext.parallelize((1 to 10).map(i => document.parse(s"{test: $i}"))) 
// 存储数据到mongodb 
mongospark.save(documents) 
// 加载数据 
val rdd = mongospark.load(spark) 
// 打印输出 
rdd.show 
} 
}

总结

以上所述是小编给大家介绍的spark整合mongodb的方法，希望对大家有所帮助