Spark介绍
按照官方的定义，Spark 是一个通用，快速，适用于大规模数据的处理引擎。
通用性：我们可以使用Spark SQL来执行常规分析， Spark Streaming 来流数据处理， 以及用Mlib来执行机器学习等。Java，python，scala及R语言的支持也是其通用性的表现之一。
快速： 这个可能是Spark成功的最初原因之一，主要归功于其基于内存的运算方式。当需要处理的数据需要反复迭代时，Spark可以直接在内存中暂存数据，而无需像Map Reduce一样需要把数据写回磁盘。官方的数据表明：它可以比传统的Map Reduce快上100倍。
大规模：原生支持HDFS，并且其计算节点支持弹性扩展，利用大量廉价计算资源并发的特点来支持大规模数据处理。
环境准备
mongodb下载
解压安装
启动mongodb服务
|
1 |
$MONGODB_HOME/bin/mongod --fork --dbpath=/root/data/mongodb/ --logpath=/root/data/log/mongodb/mongodb.log |
pom依赖
|
1 2 3 4 5 |
<dependency> <groupId>org.mongodb.spark</groupId> <artifactId>mongo-spark-connector_2. 11 </artifactId> <version>${spark.version}</version> </dependency> |
实例代码
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
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的方法，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对服务器之家网站的支持！
原文链接：http://HdhCmsTestroncoo测试数据/article/detail/131053
查看更多关于Spark整合Mongodb的方法的详细内容...