Как запросить манго с использованием искры?

Я использую искру и монго. Я могу подключиться к 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'}")

Это заняло только одно условие за раз. Я хочу добавить условие, если "использование" > 30

1) Как добавить несколько условий к запросу mongo (включая больше и меньше) с использованием искры и монго?

Также я хочу перебирать каждый документ результата запроса с помощью scala?

2) Как выполнить итерацию результата с помощью scala??

Ответы

Ответ 1

Привет, вы можете попробовать следующее:

Существует проект, который объединяет MongoDB с Spark

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

1) выполните git клон

2) зайдите внутрь глубокой искры, а затем в глубокий родительский

3) mvn install

4) открыть искровую оболочку с этими параметрами:

./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) Выполните простой пример в искровой оболочке:

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

Ответ 2

1) Чтобы добавить условия к вашему запросу, просто добавьте их в словарь, снабженный "mongo.input.query":

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

Чтобы лучше понять, как работают запросы, выполните следующие действия:

http://docs.mongodb.org/manual/tutorial/query-documents/

http://docs.mongodb.org/getting-started/python/query/

2) Для повторения результата вы можете захотеть исправить метод RDD "собирать", больше информации в этой ссылке, просто найдите метод сбора:

http://spark.apache.org/docs/latest/api/scala/#org.apache.spark.rdd.RDD