Загрузка Spark 2.x DataFrame из MongoDB GridFS в Python - PullRequest
Новая
       6

Загрузка Spark 2.x DataFrame из MongoDB GridFS в Python

1 голос
/ 16 октября 2019

Я использую pyspark sql с keras под elephas.

Я хочу попробовать какую-то распределенную обработку изображений с mongoDB GridFS

Я нашел связанный вопрос, но в мире Java наScala Загрузка фрейма данных Spark 2.x из MongoDB GridFS

Но не более того, я не могу найти никакой другой документации по работе с GridFS из pySpark.

мой pyspark - код монго выглядит так: 

sparkConf = SparkConf().setMaster("local[4]").setAppName("MongoSparkConnectorTour")\
                                             .set("spark.app.id", "MongoSparkConnectorTour")\
                                             .set("spark.mongodb.input.database", config.MONGO_DB)

# If executed via pyspark, sc is already instantiated
sc = SparkContext(conf=sparkConf)
sqlContext = SQLContext(sc)

dk = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource")\
                    .option("spark.mongodb.input.uri", config.MONGO_MED_EVENTS)\
                    .load()

if (dk.count() > 0):
    # print data frame schema
    dk.printSchema()

    # Preview Dataframe (Pandas Preview is Cleaner)
    print( dk.limit(5).toPandas() )

Возможно ли работать с данными GridFS таким образом? Я хотел бы видеть минимальный пример.

1 Ответ

1 голос
/ 18 октября 2019

Есть способ перевести код Scala в Pyspark.

  1. Скачать mongo-hadoop-core.jar из https://mvnrepository.com/artifact/org.mongodb.mongo-hadoop/mongo-hadoop-core/2.0.2

  2. Запустить pyspark с помощьюбанка в комплекте: 

SPARK_CLASSPATH=./path/to/mongo-hadoop-core.jar pyspark
И переведенный код: 
sc = SparkContext(conf=sparkConf)

mongo_conf = {
    "mongo.input.uri": "mongodb://..."
    "mongo.input.query": s"...mongo query here..."
}

rdd = sc.newAPIHadoopRDD("com.mongodb.hadoop.GridFSInputFormat", keyClass="org.apache.hadoop.io.NullWritable", valueClass="org.apache.hadoop.io.MapWritable", conf=conf)

Я не уверен на сто процентов в keyClass и valueClass, поэтому вот источники, которые я использовал для компиляции этого кода:

...