PySpark-MongoDB Connector

---

問題反饋

在使用中有任何問題，能夠反饋給我，如下聯繫方式跟我交流

• Author: Leo
• Wechat: Leo-sunhailin
• E-mail: 379978424@qq.com

---

• 目錄
  • 項目環境
  • 下載方式
  • 小問題
  • 解決辦法和過程
  • 代碼示例
  • 問題反饋

---

項目環境

• 環境部署: 請參考右邊 -> 部署步驟
• 環境詳情：
  • Spark版本: Apache Spark 2.2.0 with Hadoop2.7
  • MongoDB版本: 3.4.9-with openSSL
  • JDK版本: 1.8
  • Python版本: 3.4.4

---

下載方式

• 一、官方MongoDB-Spark Connector

# 第一種方式
spark-shell --packages org.mongodb.spark:mongo-spark-connector_2.11:2.2.0
 # 第二種
pyspark --packages org.mongodb.spark:mongo-spark-connector_2.11:2.2.0
複製代碼

• 二、第三方的Connector
  • 有點坑爹的項目項目連接
  • 項目發起者的官網打不開找不到對應的版本
  • 0.12.x(具體x有哪些不知道...)
  • 命令以下:

spark-shell --packages com.stratio.datasource:spark-mongodb_2.11:0.13.0
複製代碼

小問題:

• 問題:公司的網絡有毒,翻不出去外網(應該和公司設置的DNS有關係)

---

解決辦法和過程:

• 解決辦法:Teamviewer回家用Spark-shell下了MongoDB官方的解決方案的Jar包.(執行的是命令1)
  • 偷懶能夠下載下面的jar包,手動跳轉第四步
  • Jar包下載連接 密碼:qkkp
• 一、Maven編譯後(下載的時候就會編譯的不須要手動),生成了兩個jar包
  • 路徑默認在: C:/User/<用戶名>/.ivy2/
• 二、下載沒出錯以後在.ivy2文件下會存在兩個文件夾caches,jars.
• 三、下載完以後打開jars文件夾會有兩個jar包:
  • org.mongodb.spark_mongo-spark-connector_2.11-2.2.0.jar
  • org.mongodb_mongo-java-driver-3.4.2.jar
• 四、把兩個複製到Spark根目錄下的jars文件夾中.

---

代碼示例:

# -*- coding: UTF-8 -*-
""" Created on 2017年10月24日 @author: Leo """

import os
from pyspark.sql import SparkSession

os.environ['SPARK_HOME'] = "你的Spark根目錄"
os.environ['HADOOP_HOME'] = "你的Hadoop根目錄"


class PySparkMongoDB:
    def __init__(self):
        # 這個是uri的配置
        # mongodb://<MongoDB地址:端口>/<數據庫名>.<集合名>
        # 不指定端口就是默認27017
        self.uri_conf = "mongodb://127.0.0.1/<數據庫名>.<集合名>"
        
        # Connect MongoDB(經過SparkSession維護鏈接)
        self.my_spark = SparkSession \
            .builder \
            .appName("myApp") \
            .config("spark.mongodb.input.uri", self.uri_conf) \
            .config("spark.mongodb.output.uri", self.uri_conf) \
            .getOrCreate()

    def read_mgo(self):
        df = self.my_spark.read.format("com.mongodb.spark.sql.DefaultSource").load()
        print(df.show())


if __name__ == '__main__':
    mgo = PySparkMongoDB()
    mgo.read_mgo()
複製代碼