Spark：通过Stratio和RDD查询Mongodb

我使用Stratio（0.11。）查询MongoDB和Spark。我有兴趣使用RDD（无DataFrame）。Spark：通过Stratio和RDD查询Mongodb

我在做什么现在的问题是：

val mongoRDD = new MongodbRDD(sqlContext, readConfig, new MongodbPartitioner(readConfig)) 
mongoRDD.foreach(println)

，并显示在一个正确的方式收集的内容。

有没有办法使用查询（如String或通过QueryBuilder构建）与Stratio（在我的情况下查询是$近类型）适用于MongodbRDD？

来源

2016-02-19 Randomize

乍看之下它看起来并不像它。它将SQL谓词向下推，但它不能用于'$ near'。 – zero323

你仍然可以尝试直接使用它：https://github.com/Stratio/Spark-MongoDB/blob/231e1fcb961868963bf7efcae04778d5a9c11f78/spark-mongodb/src/main/scala/com/stratio/datasource/mongodb/rdd/MongodbRDD。 scala＃L37 – zero323

是的，我在看“过滤器”，但唯一可以工作的是用“EqualTo”构建所有查询。有点奇怪。 – Randomize

正如@ zero323已经暗示，这样做的方法是使用filters参数。这些过滤器由库进行检查，并与MongoDB QueryBuilder可用过滤器进行匹配。

从火花MongoDB的source code：

sFilters.foreach { 
    case EqualTo(attribute, value) => 
     queryBuilder.put(attribute).is(checkObjectID(attribute, value)) 
    case GreaterThan(attribute, value) => 
     queryBuilder.put(attribute).greaterThan(checkObjectID(attribute, value)) 
    case GreaterThanOrEqual(attribute, value) => 
     queryBuilder.put(attribute).greaterThanEquals(checkObjectID(attribute, value)) 
    case In(attribute, values) => 
     queryBuilder.put(attribute).in(values.map(value => checkObjectID(attribute, value))) 
    case LessThan(attribute, value) => 
     queryBuilder.put(attribute).lessThan(checkObjectID(attribute, value)) 
    case LessThanOrEqual(attribute, value) => 
     queryBuilder.put(attribute).lessThanEquals(checkObjectID(attribute, value)) 
    case IsNull(attribute) => 
     queryBuilder.put(attribute).is(null) 
    case IsNotNull(attribute) => 
     queryBuilder.put(attribute).notEquals(null) 
    case And(leftFilter, rightFilter) if !parentFilterIsNot => 
     queryBuilder.and(filtersToDBObject(Array(leftFilter)), filtersToDBObject(Array(rightFilter))) 
    case Or(leftFilter, rightFilter) if !parentFilterIsNot => 
     queryBuilder.or(filtersToDBObject(Array(leftFilter)), filtersToDBObject(Array(rightFilter))) 
    case StringStartsWith(attribute, value) if !parentFilterIsNot => 
     queryBuilder.put(attribute).regex(Pattern.compile("^" + value + ".*$")) 
    case StringEndsWith(attribute, value) if !parentFilterIsNot => 
     queryBuilder.put(attribute).regex(Pattern.compile("^.*" + value + "$")) 
    case StringContains(attribute, value) if !parentFilterIsNot => 
     queryBuilder.put(attribute).regex(Pattern.compile(".*" + value + ".*")) 
    case Not(filter) => 
     filtersToDBObject(Array(filter), true) 
    }

正如你所看到的，near不被应用，但它似乎是它可以自QueryBuilderoffers methods to use that MongoDB function很容易地添加到连接器的功能。

您可以尝试修改连接器。不过，我会尽力实施它，并在接下来的日子里做一个PR。

编辑：

一个PR has been opened包括描述$near所以你可以使用MongodbRdd作为源过滤器类型：

val mongoRDD = new MongodbRDD(
    sqlContext, 
    readConfig, 
    new MongodbPartitioner(readConfig), 
    filters = FilterSection(Array(Near("x", 3.0, 4.0)))) 
)

来源

2016-02-19 17:30:39

是的，我已经看过GitHub的那部分代码。理想的情况是，只需要有一个“setQuery”方法，可以将查询生成的查询通过QueryBuilder或文本字符串正常传递。 – Randomize

@Randomize这是可行的。然而，它更容易添加“近”过滤器 –

是的，我同意（虽然有近不同类型的查询）。顺便说一句，他们正在使用的抽象是将代码耦合到特定版本的mongodb-java-driver，因为QueryBuilder已经将API从2. *更改为3. *。 – Randomize

Spark：通过Stratio和RDD查询Mongodb

回答

相关问题