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

作者簡介

張友東,阿里云高級技術專家,主要關注分布式存儲與數據庫等技術領域,目前參與阿里云數據庫(PolarDB、MySQL、MongoDB、Redis)等項目的開發工作,致力于讓開發者用上最好的云數據庫服務。

發表評論