Spark整合Mongodb的方法_MongoDB

Spark整合Mongodb的方法

2020-05-16 18:59mrr MongoDB

Spark 是一个通用，快速，适用于大规模数据的处理引擎。接下来通过本文给大家分享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 

									} 

									}