Why Spark with MongoDB?

  1. 高性能，官方号称 100x faster，因为可以全内存运行，性能提升肯定是很明显的
  2. 简单易用，支持 Java、Python、Scala、SQL 等多种语言，使得构建分析应用非常简单
  3. 统一构建 ，支持多种数据源，通过 Spark RDD 屏蔽底层数据差异，同一个分析应用可运行于不同的数据源；
  4. 应用场景广泛，能同时支持批处理以及流式处理

MongoDB Spark Connector 为官方推出，用于适配 Spark 操作 MongoDB 数据；本文以 Python 为例，介绍 MongoDB Spark Connector 的使用，帮助你基于 MongoDB 构建第一个分析应用。

准备 MongoDB 环境

安装 MongoDB 参考 Install MongoDB Community Edition on Linux


1. mkdir mongodata
2. mongod --dbpath mongodata --port 9555

准备 Spark python 环境

参考 PySpark - Quick Guide

下载 Spark


1. cd /home/mongo-spark
2. wget http://mirrors.tuna.tsinghua.edu.cn/apache/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz
3. tar zxvf spark-2.4.4-bin-hadoop2.7.tgz

设置 Spark 环境变量


1. export SPARK_HOME=/home/mongo-spark/spark-2.4.4-bin-hadoop2.7
2. export PATH=$PATH:/home/mongo-spark/spark-2.4.4-bin-hadoop2.7/bin
3. export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.4-src.zip:$PYTHONPATH
4. export PATH=$SPARK_HOME/python:$PATH

运行 Spark RDD 示例


1. # count.py
2. from pyspark import SparkContext
3. sc = SparkContext("local", "count app")
4. words = sc.parallelize (
5. ["scala",
6. "java",
7. "hadoop",
8. "spark",
9. "akka",
10. "spark vs hadoop",
11. "pyspark",
12. "pyspark and spark"]
13. )
14. counts = words.count()

16. $SPARK_HOME/bin/spark-submit count.py
17. Number of elements in RDD → 8

如果上述程序运行成功，说明 Spark python 环境准备成功，还可以测试 Spark 的其他 RDD 操作，比如 collector、filter、map、reduce、join 等，更多示例参考 PySpark - Quick Guide

Spark 操作 MongoDB 数据

参考 Spark Connector Python Guide

准备测试数据 test.coll01 插入3条测试数据，test.coll02 未空


1. mongo --port 9555

3. > db.coll01.find()
4. { "_id" : 1, "type" : "apple", "qty" : 5 }
5. { "_id" : 2, "type" : "orange", "qty" : 10 }
6. { "_id" : 3, "type" : "banana", "qty" : 15 }
7. > db.coll02.find()

准备操作脚本，将输入集合的数据按条件进行过滤，写到输出集合


1. # mongo-spark-test.py
2. from pyspark.sql import SparkSession

4. # Create Spark Session

6. spark = SparkSession \
7. .builder \
8. .appName("myApp") \
9. .config("spark.mongodb.input.uri", "mongodb://127.0.0.1:9555/test.coll01") \
10. .config("spark.mongodb.output.uri", "mongodb://127.0.0.1:9555/test.coll") \
11. .getOrCreate()

14. # Read from MongoDB
15. df = spark.read.format("mongo").load()
16. df.show()

18. # Filter and Write
19. df.filter(df['qty'] >= 10).write.format("mongo").mode("append").save()

21. # Use SQL
22. # df.createOrReplaceTempView("temp")
23. # some_fruit = spark.sql("SELECT type, qty FROM temp WHERE type LIKE '%e%'")
24. # some_fruit.show()

运行脚本


1. $SPARK_HOME/bin/spark-submit --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.1 mongo-spark-test.py

3. mongo --port 9555

5. > db.coll02.find()
6. { "_id" : 2, "qty" : 10, "type" : "orange" }
7. { "_id" : 3, "qty" : 15, "type" : "banana" }

原文：http://mysql.taobao.org/monthly/2019/10/07/