MongoDB Spark Connector 實戰指南

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

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" }