Wie erstellt Spark Datenrahmen mit gefilterten Datensätzen aus MongoDB?

Question

Meine Anwendung wurde mit MongoDB als Plattform erstellt. Eine Sammlung in DB hat ein riesiges Datenvolumen und hat sich für apache spark entschieden, um analytische Daten durch Berechnung zu erhalten und zu generieren. Ich habe Spark Connector for MongoDB konfiguriert, um mit MongoDB zu kommunizieren. Ich muss MongoDB-Sammlung unter Verwendung pyspark abfragen und einen Datenrahmen aufbauen, der resultset der mongodb Frage besteht. Bitte schlagen Sie mir eine geeignete Lösung vor.

Answer 1

Sie können wie so die Daten direkt in einen Datenrahmen laden:

# Create the dataframe 
df = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource").option("uri", "mongodb://127.0.0.1/mydb.mycoll").load() 

# Filter the data via the api 
df.filter(people.age > 30) 

# Filter via sql 
df.registerTempTable("people") 
over_thirty = sqlContext.sql("SELECT name, age FROM people WHERE age > 30") 

Für weitere Informationen, um den Mongo Spark-Anschluss Python API Abschnitt oder die introduction.py sehen. Die SQL-Abfragen werden übersetzt und an den Connector zurückgegeben, sodass die Daten in MongoDB abgefragt werden können, bevor sie an den Funke-Cluster gesendet werden.

Sie können auch Ihre eigenen bieten aggregation pipeline auf die Sammlung anzuwenden, bevor die Ergebnisse in Spark-Rückkehr:

dfr = sqlContext.read.option("pipeline", "[{ $match: { name: { $exists: true } } }]") 
df = dfr.option("uri", ...).format("com.mongodb.spark.sql.DefaultSource").load()