MongoDB Spark Connector 实战指南

2021-11-02 07:12:20

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
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 等，更多买QQ示例参考 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