python - 使Spark，Python和MongoDB协同工作

Question

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

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

Spark spark-1.5.1-bin-hadoop2.6

Hadoop hadoop-2.6.1

蒙戈 2.6.10

Mongo-Hadoop连接器从https://github.com/mongodb/mongo-hadoop.git克隆

Python 2.7.10

我在执行各个步骤时遇到了一些困难，例如将哪些 jar 添加到哪个路径，所以我添加的是
/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

根据 here

This exception is raised when an exception occurs in the Java client code. For example, if you try to pop an element from an empty stack. The instance of the Java exception thrown is stored in the java_exception member.

查看 pymongo_spark.py的源代码和引发错误的行，它说

"Error while communicating with the JVM. Is the MongoDB Spark jar on Spark's CLASSPATH? : "

因此，作为回应，我试图确保传递了正确的 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，并且可以。

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

Answer 1

更新:

2016-07-04

自上次更新以来，MongoDB Spark Connector相当成熟。它提供了up-to-date binaries和基于数据源的API，但它使用的是SparkConf配置，因此，它在主观上不如Stratio / Spark-MongoDB灵活。

2016-03-30

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

mongodb/mongo-spark

Stratio/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更加稳定，支持谓词下推而无需静态配置，并且可以正常工作。

原始答案:

确实，这里有很多 Activity 的部分。我试图通过构建一个与所描述的配置大致匹配的简单Docker镜像，使其更易于管理(不过，为简洁起见，我省略了Hadoop库)。您可以找到 complete source on 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 .

或下载我拥有 pushed to 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

通过 --jars和 --driver-class-path启动PySpark shell:

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似乎关闭了连接。因此，在collect之后调用 rdd.count()例如将引发异常。

基于创建此图像时遇到的不同问题，我倾向于认为 将 mongo-hadoop-1.5.0-SNAPSHOT.jar和 mongo-hadoop-spark-1.5.0-SNAPSHOT.jar 传递给 --jars和 --driver-class-path 是唯一的要求。

注释:

此图像是基于jaceklaskowski/docker-spark的，因此请确保向@jacek-laskowski发送一些良好的业障。

如果不需要包括new API的开发版本，那么使用--packages最有可能是一个更好的选择。