无法使用spark-shell从EMR集群连接到远程MongoDB

Question

我正在尝试从EMR集群连接到远程Mongo数据库。使用命令spark-shell --packages com.stratio.datasource:spark-mongodb_2.10:0.11.2执行以下代码

import com.stratio.datasource.mongodb._
import com.stratio.datasource.mongodb.config._
import com.stratio.datasource.mongodb.config.MongodbConfig._

val builder = MongodbConfigBuilder(Map(Host -> List("[IP.OF.REMOTE.HOST]:3001"), Database -> "meteor", Collection ->"my_target_collection", ("user", "user_name"), ("database", "meteor"), ("password", "my_password")))
val readConfig = builder.build()
val mongoRDD = sqlContext.fromMongoDB(readConfig)

Spark-shell响应以下错误：

16/07/26 15:44:35 INFO SparkContext: Starting job: aggregate at MongodbSchema.scala:47
16/07/26 15:44:45 WARN DAGScheduler: Creating new stage failed due to exception - job: 1
com.mongodb.MongoTimeoutException: Timed out after 10000 ms while waiting to connect. Client view of cluster state is {type=Unknown, servers=[{address=[IP.OF.REMOTE.HOST]:3001, type=Unknown, state=Connecting, exception={java.lang.IllegalArgumentException: response too long: 1347703880}}]
    at com.mongodb.BaseCluster.getDescription(BaseCluster.java:128)
    at com.mongodb.DBTCPConnector.getClusterDescription(DBTCPConnector.java:394)
    at com.mongodb.DBTCPConnector.getType(DBTCPConnector.java:571)
    at com.mongodb.DBTCPConnector.getReplicaSetStatus(DBTCPConnector.java:362)
    at com.mongodb.Mongo.getReplicaSetStatus(Mongo.java:446)
.
.
.

在阅读了一段时间后，SO和其他论坛上的一些回复指出，java.lang.IllegalArgumentException: response too long: 1347703880错误可能是由错误的Mongo驱动程序引起的。基于此，我开始使用更新的驱动程序执行spark-shell，如下所示：

spark-shell --packages com.stratio.datasource:spark-mongodb_2.10:0.11.2 --jars casbah-commons_2.10-3.1.1.jar,casbah-core_2.10-3.1.1.jar,casbah-query_2.10-3.1.1ja.jar,mongo-java-driver-2.13.0.jar

当然，在此之前，我下载了jars并以与执行spark-shell相同的方式存储它们。尽管如此，对于这种方法，spark-shell会给出以下隐含的错误消息：

Exception in thread "dag-scheduler-event-loop" java.lang.NoClassDefFoundError: com/mongodb/casbah/query/dsl/CurrentDateOp
    at com.mongodb.casbah.MongoClient.apply(MongoClient.scala:218)
    at com.stratio.datasource.mongodb.partitioner.MongodbPartitioner.isShardedCollection(MongodbPartitioner.scala:78)

值得一提的是，目标MongoDB是Meteor Mongo数据库，这就是为什么我尝试使用[IP.OF.REMOTE.HOST]:3001连接，而不是使用端口27017。

可能的问题是什么？我遵循了许多教程，但似乎所有教程都在同一个主机中具有MongoDB，允许它们在凭据中声明localhost:27017。我是不是漏掉了什么？

谢谢你的帮助！

Answer 1

我最终转而使用MongoDB的官方Java驱动程序。这是我第一次使用Spark和Scala编程语言，所以我还不太熟悉使用纯Java JAR的想法。

解决方案

我下载了必要的JAR，并将它们存储在作业文件所在的目录中，这是一个Scala文件。所以目录看起来像这样：

/job_directory
|--job.scala
|--bson-3.0.1.jar
|--mongodb-driver-3.0.1.jar
|--mongodb-driver-core-3.0.1.jar

然后，我启动spark-shell，如下所示，将JAR及其类加载到shell环境中：

spark-shell --jars "mongodb-driver-3.0.1.jar,mongodb-driver-core-3.0.1.jar,bson-3.0.1.jar"

接下来，我执行以下命令，将作业的源代码加载到spark-shell中：

:load job.scala

最后，我在作业中执行main对象，如下所示：

MainObject.main(Array())

作为MainObject中的代码，它只是像tutorial所说的那样：

val mongo = new MongoClient(IP_OF_REMOTE_MONGO , 27017)
val db = mongo.getDB(DB_NAME)

希望这能帮助未来的读者和spark-shell/Scala初学者！