无法使用 spark-shell 从 EMR 集群连接到远程 MongoDB
Cannot connect to remote MongoDB from EMR cluster with spark-shell
我正在尝试从 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
当然，在此之前我下载了 jar 并将它们存储在与执行 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 而不是使用端口 [=19] =].
可能是什么问题？我遵循了许多教程，但它们似乎都在同一主机中拥有 MongoDB，允许它们在凭据中声明 localhost:27017。有什么我想念的吗？
感谢您的帮助！
我最终改用 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
最后我像这样在我的工作中执行主要对象：
MainObject.main(Array())
MainObject 中的代码，仅如 tutorial 所述：
val mongo = new MongoClient(IP_OF_REMOTE_MONGO , 27017)
val db = mongo.getDB(DB_NAME)
希望这对未来的读者和 spark-shell/Scala 初学者有所帮助！
我正在尝试从 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
当然，在此之前我下载了 jar 并将它们存储在与执行 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 而不是使用端口 [=19] =].
可能是什么问题？我遵循了许多教程，但它们似乎都在同一主机中拥有 MongoDB，允许它们在凭据中声明 localhost:27017。有什么我想念的吗？
感谢您的帮助！
我最终改用 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
最后我像这样在我的工作中执行主要对象：
MainObject.main(Array())
MainObject 中的代码，仅如 tutorial 所述：
val mongo = new MongoClient(IP_OF_REMOTE_MONGO , 27017)
val db = mongo.getDB(DB_NAME)
希望这对未来的读者和 spark-shell/Scala 初学者有所帮助！