Fazendo Spark, Python e MongoDB trabalharem juntos

Atualizações :

2016-07-04

Desde a última atualização, o MongoDB Spark Connector amadureceu bastante. Ele fornece binários atualizados e API baseada em fonte de dados, mas está usando o SparkConf configuração por isso é subjetivamente menos flexível que o Stratio/Spark-MongoDB.

30-03-2016

Desde a resposta original, encontrei duas maneiras diferentes de se conectar ao MongoDB do Spark:

mongodb/mongo-spark
Stratio/Spark-MongoDB

Enquanto o primeiro parece ser relativamente imaturo, o último parece uma escolha muito melhor do que um conector Mongo-Hadoop e fornece uma API Spark SQL.

# Adjust Scala and package version according to your setup
# although officially 0.11 supports only Spark 1.5
# I haven't encountered any issues on 1.6.1
bin/pyspark --packages com.stratio.datasource:spark-mongodb_2.11:0.11.0

df = (sqlContext.read
  .format("com.stratio.datasource.mongodb")
  .options(host="mongo:27017", database="foo", collection="bar")
  .load())

df.show()

## +---+----+--------------------+
## |  x|   y|                 _id|
## +---+----+--------------------+
## |1.0|-1.0|56fbe6f6e4120712c...|
## |0.0| 4.0|56fbe701e4120712c...|
## +---+----+--------------------+

Parece ser muito mais estável que mongo-hadoop-spark , suporta pushdown de predicado sem configuração estática e simplesmente funciona.

A resposta original :

Na verdade, existem algumas partes móveis aqui. Tentei torná-lo um pouco mais gerenciável criando uma imagem simples do Docker que corresponde aproximadamente à configuração descrita (embora omiti as bibliotecas do Hadoop por brevidade). Você pode encontrar a fonte completa no GitHub (DOI 10.5281/zenodo.47882) e construa do zero:

git clone https://github.com/zero323/docker-mongo-spark.git
cd docker-mongo-spark
docker build -t zero323/mongo-spark .

ou baixe uma imagem que enviei para o Docker Hub para que você possa simplesmente docker pull zero323/mongo-spark ):

Iniciar imagens:

docker run -d --name mongo mongo:2.6
docker run -i -t --link mongo:mongo zero323/mongo-spark /bin/bash

Inicie o shell do PySpark passando --jars e --driver-class-path :

pyspark --jars ${JARS} --driver-class-path ${SPARK_DRIVER_EXTRA_CLASSPATH}

E por fim veja como funciona:

import pymongo
import pymongo_spark

mongo_url = 'mongodb://mongo:27017/'

client = pymongo.MongoClient(mongo_url)
client.foo.bar.insert_many([
    {"x": 1.0, "y": -1.0}, {"x": 0.0, "y": 4.0}])
client.close()

pymongo_spark.activate()
rdd = (sc.mongoRDD('{0}foo.bar'.format(mongo_url))
    .map(lambda doc: (doc.get('x'), doc.get('y'))))
rdd.collect()

## [(1.0, -1.0), (0.0, 4.0)]

Observe que o mongo-hadoop parece fechar a conexão após a primeira ação. Então, chamando por exemplo rdd.count() após a coleta lançará uma exceção.

Com base em diferentes problemas que encontrei ao criar esta imagem, costumo acreditar que passar mongo-hadoop-1.5.0-SNAPSHOT.jar e mongo-hadoop-spark-1.5.0-SNAPSHOT.jar para ambos --jars e --driver-class-path é o único requisito difícil .

Observações :

Esta imagem é vagamente baseada em jaceklaskowski/docker-spark, então envie um bom karma para @jacek-laskowski se isso ajudar.
Se não exigir uma versão de desenvolvimento incluindo uma nova API, use --packages é provavelmente uma opção melhor.