1. 首页
  2. 问题
  3. 使Spark，Python和MongoDB协同工作
小编典典

使Spark，Python和MongoDB协同工作

python

我很难让这些组件正确地编织在一起。我已经安装了Spark并且可以成功工作，我可以在本地，独立运行作业，也可以通过YARN运行作业。据我所知，我已经按照此处和此处建议的步骤进行操作

我正在使用Ubuntu，而我拥有的各种组件版本是

我在执行各个步骤时遇到了一些困难，例如将哪些罐子添加到哪个路径，所以我添加的是

  • 在/usr/local/share/hadoop-2.6.1/share/hadoop/mapreduce 我添加 mongo-hadoop-core-1.5.0-SNAPSHOT.jar
  • 以下 环境变量
    • export HADOOP_HOME="/usr/local/share/hadoop-2.6.1"
    • export PATH=$PATH:$HADOOP_HOME/bin
    • export SPARK_HOME="/usr/local/share/spark-1.5.1-bin-hadoop2.6"
    • export PYTHONPATH="/usr/local/share/mongo-hadoop/spark/src/main/python"
    • export PATH=$PATH:$SPARK_HOME/bin

我的Python程序是基本的

from pyspark import SparkContext, SparkConf
import pymongo_spark
pymongo_spark.activate()

def main():
    conf = SparkConf().setAppName("pyspark test")
    sc = SparkContext(conf=conf)
    rdd = sc.mongoRDD(
        'mongodb://username:password@localhost:27017/mydb.mycollection')

if __name__ == '__main__':
    main()

我正在使用命令运行它

$SPARK_HOME/bin/spark-submit --driver-class-path /usr/local/share/mongo-hadoop/spark/build/libs/ --master local[4] ~/sparkPythonExample/SparkPythonExample.py

结果我得到以下输出

Traceback (most recent call last):
  File "/home/me/sparkPythonExample/SparkPythonExample.py", line 24, in <module>
    main()
  File "/home/me/sparkPythonExample/SparkPythonExample.py", line 17, in main
    rdd = sc.mongoRDD('mongodb://username:password@localhost:27017/mydb.mycollection')
  File "/usr/local/share/mongo-hadoop/spark/src/main/python/pymongo_spark.py", line 161, in mongoRDD
    return self.mongoPairRDD(connection_string, config).values()
  File "/usr/local/share/mongo-hadoop/spark/src/main/python/pymongo_spark.py", line 143, in mongoPairRDD
    _ensure_pickles(self)
  File "/usr/local/share/mongo-hadoop/spark/src/main/python/pymongo_spark.py", line 80, in _ensure_pickles
    orig_tb)
py4j.protocol.Py4JError

根据这里

当Java客户端代码中发生异常时，将引发此异常。例如，如果您尝试从空堆栈中弹出一个元素。引发的Java异常的实例存储在java_exception成员中。

查看源代码pymongo_spark.py并抛出错误，它说

“与JVM通信时出错。Spark的CLASSPATH上的MongoDB Spark jar是吗？：”

因此，作为回应，我试图确保传递了正确的罐子，但是我可能做错了所有事情，请参见下文

$SPARK_HOME/bin/spark-submit --jars /usr/local/share/spark-1.5.1-bin-hadoop2.6/lib/mongo-hadoop-spark-1.5.0-SNAPSHOT.jar,/usr/local/share/spark-1.5.1-bin-hadoop2.6/lib/mongo-java-driver-3.0.4.jar --driver-class-path /usr/local/share/spark-1.5.1-bin-hadoop2.6/lib/mongo-java-driver-3.0.4.jar,/usr/local/share/spark-1.5.1-bin-hadoop2.6/lib/mongo-hadoop-spark-1.5.0-SNAPSHOT.jar --master local[4] ~/sparkPythonExample/SparkPythonExample.py

我已经导入pymongo到相同的python程序中，以验证我至少可以使用它访问MongoDB，并且可以。

我知道这里有很多活动部件，所以如果我能提供更多有用的信息，请告诉我。


阅读 221

收藏
2020-12-20

共1个答案

小编典典

更新内容 ：

2016-07-04

自上次更新以来，MongoDB Spark Connector已经相当成熟。它提供了最新的二进制文件和基于数据源的API，但它使用的是SparkConf配置，因此从主观上讲它不如Stratio
/ Spark-MongoDB灵活。

2016-03-30

从原始答案开始，我发现了两种从Spark连接到MongoDB的不同方法：

尽管前者似乎还不成熟，但后者似乎比Mongo-Hadoop连接器更好，并且提供了Spark SQL API。

# Adjust Scala and package version according to your setup
# although officially 0.11 supports only Spark 1.5
# I haven't encountered any issues on 1.6.1
bin/pyspark --packages com.stratio.datasource:spark-mongodb_2.11:0.11.0



df = (sqlContext.read
  .format("com.stratio.datasource.mongodb")
  .options(host="mongo:27017", database="foo", collection="bar")
  .load())

df.show()

## +---+----+--------------------+
## |  x|   y|                 _id|
## +---+----+--------------------+
## |1.0|-1.0|56fbe6f6e4120712c...|
## |0.0| 4.0|56fbe701e4120712c...|
## +---+----+--------------------+

它似乎比稳定得多mongo-hadoop-spark，无需静态配置即可支持谓词下推，并且可以正常工作。

原始答案 ：

确实，这里有很多活动部件。我试图通过构建一个与描述的配置大致匹配的简单Docker映像，使其更易于管理（不过，为简洁起见，我省略了Hadoop库）。您可以在GitHub（DOI 10.5281 /
zenodo.47882）上找到完整的源代码，并从头开始构建它：

git clone https://github.com/zero323/docker-mongo-spark.git
cd docker-mongo-spark
docker build -t zero323/mongo-spark .

或下载我已推送到Docker Hub的映像，因此您可以简单地docker pull zero323/mongo-spark：

起始图片：

docker run -d --name mongo mongo:2.6
docker run -i -t --link mongo:mongo zero323/mongo-spark /bin/bash

启动PySpark shell传递--jars并--driver-class-path：

pyspark --jars ${JARS} --driver-class-path ${SPARK_DRIVER_EXTRA_CLASSPATH}

最后看看它是如何工作的：

import pymongo
import pymongo_spark

mongo_url = 'mongodb://mongo:27017/'

client = pymongo.MongoClient(mongo_url)
client.foo.bar.insert_many([
    {"x": 1.0, "y": -1.0}, {"x": 0.0, "y": 4.0}])
client.close()

pymongo_spark.activate()
rdd = (sc.mongoRDD('{0}foo.bar'.format(mongo_url))
    .map(lambda doc: (doc.get('x'), doc.get('y'))))
rdd.collect()

## [(1.0, -1.0), (0.0, 4.0)]

请注意，在第一个操作之后，mongo-hadoop似乎关闭了连接。因此，例如rdd.count()在collect之后调用将引发异常。

根据不同的问题，我以前遇到过这种创建图片我倾向于相信， 通过 mongo-hadoop-1.5.0-SNAPSHOT.jar与mongo- hadoop-spark-1.5.0-SNAPSHOT.jar 这两个 --jars和--driver-class-path
是唯一的硬性要求 。

注意事项 ：

2020-12-20