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()
复制代码