如何使用spark查询mongo？

Question

我正在使用spark和mongo.我可以使用以下代码连接到mongo:

val sc = new SparkContext("local", "Hello from scala")

val config = new Configuration()
config.set("mongo.input.uri", "mongodb://127.0.0.1:27017/dbName.collectionName")
val mongoRDD = sc.newAPIHadoopRDD(config, classOf[com.mongodb.hadoop.MongoInputFormat], classOf[Object], classOf[BSONObject])

上面的代码给了我收集的所有文件.

现在我想在查询中应用一些条件.

为此我用过

config.set("mongo.input.query","{customerId: 'some mongo id'}")

这一次只涉及一个条件.如果'usage'> 30,我想添加一个条件

1)如何使用spark和mongo为mongo查询添加多个条件(包括大于和小于)？

另外我想用scala迭代查询结果的每个文件？

2)如何使用scala迭代结果？

Answer 1

嗨你可以试试这个:

有一个项目将MongoDB与Spark集成在一起

https://github.com/Stratio/deep-spark/tree/develop

1)做一个git克隆

2)进入深火花,然后进入深亲

3)mvn安装

4)使用以下选项打开spark-shell:

./spark-shell --jars YOUR_PATH/deep-core-0.7.0-SNAPSHOT.jar,YOUR_PATH/deep-commons-0.7.0-SNAPSHOT.jar,YOUR_PATH/deep-mongodb-0.7.0-SNAPSHOT.jar, YOUR_PATH /蒙戈的Java驱动程序 - 2.12.4-sources.jar

记得用真实路径覆盖"YOUR_PATH"

5)在spark shell中执行一个简单的例子:

import com.stratio.deep.mongodb.config.MongoDeepJobConfig
import com.stratio.deep.mongodb.extractor.MongoNativeDBObjectExtractor
import com.stratio.deep.core.context.DeepSparkContext
import com.mongodb.DBObject
import org.apache.spark.rdd.RDD
import com.mongodb.QueryBuilder
import com.mongodb.BasicDBObject

val host = "localhost:27017"


val database = "test"

val inputCollection = "input";

val deepContext: DeepSparkContext = new DeepSparkContext(sc)

val inputConfigEntity: MongoDeepJobConfig[DBObject] = new MongoDeepJobConfig[DBObject](classOf[DBObject])


val query: QueryBuilder  = QueryBuilder.start();

query.and("number").greaterThan(27).lessThan(30);


inputConfigEntity.host(host).database(database).collection(inputCollection).filterQuery(query).setExtractorImplClass(classOf[MongoNativeDBObjectExtractor])


val inputRDDEntity: RDD[DBObject] = deepContext.createRDD(inputConfigEntity)

最好的是你可以使用QueryBuilder对象来进行查询

你也可以像这样传递一个DBObject:

{ "number" : { "$gt" : 27 , "$lt" : 30}}

如果要迭代,可以使用yourRDD.collect()方法.您也可以使用您的RDD.foreach,但您必须提供一个功能.

还有另一种方法可以将罐子添加到火花中.您可以修改spark-env.sh并将此行放在最后:

CONFDIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
for jar in $(ls $CONFDIR/../lib/*.jar); do
  SPARK_CLASSPATH=$SPARK_CLASSPATH:${jar}
done

在lib文件夹中你放了你的库,就是这样.

免责声明:我目前正在研究Stratio