兴迈手把手教你 | MongoDB Spark Connector 实战指南
Why Spark with MongoDB?
- 高性能,官方号称 100x faster,因为可以全内存运行,性能提升肯定是很明显的
- 简单易用,支持 Java、Python、Scala、SQL 等多种语言,使得构建分析应用非常简单
- 统一构建 ,支持多种数据源,通过 Spark RDD 屏蔽底层数据差异,同一个分析应用可运行于不同的数据源;
- 应用场景广泛,能同时支持批处理以及流式处理
MongoDB Spark Connector 为官方推出,用于适配 Spark 操作 MongoDB 数据;本文以 Python 为例,介绍 MongoDB Spark Connector 的使用,帮助你基于 MongoDB 构建第一个分析应用。
准备 MongoDB 环境
安装 MongoDB 参考 Install MongoDB Community Edition on Linux
mkdir mongodata mongod --dbpath mongodata --port 9555
准备 Spark python 环境
参考 PySpark - Quick Guide
下载 Spark
cd /home/mongo-spark wget http://mirrors.tuna.tsinghua.edu.cn/apache/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz tar zxvf spark-2.4.4-bin-hadoop2.7.tgz
设置 Spark 环境变量
export SPARK_HOME=/home/mongo-spark/spark-2.4.4-bin-hadoop2.7 export PATH=$PATH:/home/mongo-spark/spark-2.4.4-bin-hadoop2.7/bin export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.4-src.zip:$PYTHONPATH export PATH=$SPARK_HOME/python:$PATH
运行 Spark RDD 示例
# count.py from pyspark import SparkContext sc = SparkContext("local", "count app") words = sc.parallelize ( ["scala", "java", "hadoop", "spark", "akka", "spark vs hadoop", "pyspark", "pyspark and spark"] ) counts = words.count() $SPARK_HOME/bin/spark-submit count.py 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 未空
mongo --port 9555 > db.coll01.find() { "_id" : 1, "type" : "apple", "qty" : 5 } { "_id" : 2, "type" : "orange", "qty" : 10 } { "_id" : 3, "type" : "banana", "qty" : 15 } > db.coll02.find()
准备操作脚本,将输入集合的数据按条件进行过滤,写到输出集合
# mongo-spark-test.py from pyspark.sql import SparkSession # Create Spark Session spark = SparkSession \ .builder \ .appName("myApp") \ .config("spark.mongodb.input.uri", "mongodb://127.0.0.1:9555/test.coll01") \ .config("spark.mongodb.output.uri", "mongodb://127.0.0.1:9555/test.coll") \ .getOrCreate() # Read from MongoDB df = spark.read.format("mongo").load() df.show() # Filter and Write df.filter(df['qty'] >= 10).write.format("mongo").mode("append").save() # Use SQL # df.createOrReplaceTempView("temp") # some_fruit = spark.sql("SELECT type, qty FROM temp WHERE type LIKE '%e%'") # some_fruit.show()
运行脚本
$SPARK_HOME/bin/spark-submit --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.1 mongo-spark-test.py mongo --port 9555 > db.coll02.find() { "_id" : 2, "qty" : 10, "type" : "orange" } { "_id" : 3, "qty" : 15, "type" : "banana" }
发表评论 取消回复