sparkmongojava_java操作spark读写mongodb

首先要引入mongodb-spark-connector的maven依赖&＃xff0c;具体的可见这个api网址&＃xff1a;https://docs.mongodb.com/spark-connector/current/java-api/&＃xff0c;然后基本上就可以按照api上面的内容来进行spark操作了。这里面已经有spark读入mongodb数据转化为rdd的操作了。

有一些补充的或许有用(?)的代码&＃xff0c;放在这里。

import com.mongodb.MongoClient;

import com.mongodb.MongoClientURI;

import com.mongodb.client.MongoDatabase;

import com.mongodb.spark.MongoConnector;

import com.mongodb.spark.MongoSpark;

import com.mongodb.spark.config.ReadConfig;

import com.mongodb.spark.config.WriteConfig;

import com.mongodb.spark.rdd.api.java.JavaMongoRDD;

import com.mongodb.spark.sql.helpers.StructFields;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.Function;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import org.apache.spark.sql.types.DataTypes;

import org.apache.spark.sql.types.StructType;

import org.bson.Document;

import org.bson.types.ObjectId;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import static java.lang.String.format;

import static java.util.Arrays.asList;

import static java.util.Collections.singletonList;

public final class JavaIntroduction {

/**

* Run this main method to see the output of this quick example.

*

* &＃64;param args takes an optional single argument for the connection string

* &＃64;throws InterruptedException if a latch is interrupted

*/

public static void main(final String[] args) throws InterruptedException {

JavaSparkContext jsc &＃61; createJavaSparkContext(args);

// Create a RDD

JavaRDD documents &＃61; jsc.parallelize(asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).map

(new Function() {

&＃64;Override

public Document call(final Integer i) throws Exception {

return Document.parse("{test: " &＃43; i &＃43; "}");

}

});

// Saving data from an RDD to MongoDB

MongoSpark.save(documents);

// Saving data with a custom WriteConfig

Map writeOverrides &＃61; new HashMap();

writeOverrides.put("collection", "spark");

writeOverrides.put("writeConcern.w", "majority");

WriteConfig writeConfig &＃61; WriteConfig.create(jsc).withOptions(writeOverrides);

JavaRDD sparkDocuments &＃61; jsc.parallelize(asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).map

(new Function() {

&＃64;Override

public Document call(final Integer i) throws Exception {

return Document.parse("{spark: " &＃43; i &＃43; "}");

}

});

// Saving data from an RDD to MongoDB

MongoSpark.save(sparkDocuments, writeConfig);

// Loading and analyzing data from MongoDB

JavaMongoRDD rdd &＃61; MongoSpark.load(jsc);

System.out.println(rdd.count());

System.out.println(rdd.first().toJson());

// Loading data with a custom ReadConfig

Map readOverrides &＃61; new HashMap();

readOverrides.put("collection", "spark");

readOverrides.put("readPreference.name", "secondaryPreferred");

ReadConfig readConfig &＃61; ReadConfig.create(jsc).withOptions(readOverrides);

JavaMongoRDD customRdd &＃61; MongoSpark.load(jsc, readConfig);

System.out.println(customRdd.count());

System.out.println(customRdd.first().toJson());

// Filtering an rdd using an aggregation pipeline before passing data to Spark

JavaMongoRDD aggregatedRdd &＃61; rdd.withPipeline(singletonList(Document.parse("{ $match: { test : { $gt : 5 } } }")));

System.out.println(aggregatedRdd.count());

System.out.println(aggregatedRdd.first().toJson());

// Datasets

// Drop database

dropDatabase(getMongoClientURI(args));

// Add Sample Data

List characters &＃61; asList(

"{&＃39;name&＃39;: &＃39;Bilbo Baggins&＃39;, &＃39;age&＃39;: 50}",

"{&＃39;name&＃39;: &＃39;Gandalf&＃39;, &＃39;age&＃39;: 1000}",

"{&＃39;name&＃39;: &＃39;Thorin&＃39;, &＃39;age&＃39;: 195}",

"{&＃39;name&＃39;: &＃39;Balin&＃39;, &＃39;age&＃39;: 178}",

"{&＃39;name&＃39;: &＃39;K铆li&＃39;, &＃39;age&＃39;: 77}",

"{&＃39;name&＃39;: &＃39;Dwalin&＃39;, &＃39;age&＃39;: 169}",

"{&＃39;name&＃39;: &＃39;脫in&＃39;, &＃39;age&＃39;: 167}",

"{&＃39;name&＃39;: &＃39;Gl贸in&＃39;, &＃39;age&＃39;: 158}",

"{&＃39;name&＃39;: &＃39;F铆li&＃39;, &＃39;age&＃39;: 82}",

"{&＃39;name&＃39;: &＃39;Bombur&＃39;}"

);

MongoSpark.save(jsc.parallelize(characters).map(new Function() {

&＃64;Override

public Document call(final String json) throws Exception {

return Document.parse(json);

}

}));

// Load inferring schema

Dataset df &＃61; MongoSpark.load(jsc).toDF();

df.printSchema();

df.show();

// Declare the Schema via a Java Bean

SparkSession sparkSession &＃61; SparkSession.builder().getOrCreate();

Dataset explicitDF &＃61; MongoSpark.load(jsc).toDF(Character.class);

explicitDF.printSchema();

// SQL

explicitDF.registerTempTable("characters");

Dataset centenarians &＃61; sparkSession.sql("SELECT name, age FROM characters WHERE age >&＃61; 100");

// Saving DataFrame

MongoSpark.write(centenarians).option("collection", "hundredClub").save();

MongoSpark.load(sparkSession, ReadConfig.create(sparkSession).withOption("collection", "hundredClub"), Character.class).show();

// Drop database

MongoConnector.apply(jsc.sc()).withDatabaseDo(ReadConfig.create(sparkSession), new Function() {

&＃64;Override

public Void call(final MongoDatabase db) throws Exception {

db.drop();

return null;

}

});

String objectId &＃61; "123400000000000000000000";

List docs &＃61; asList(

new Document("_id", new ObjectId(objectId)).append("a", 1),

new Document("_id", new ObjectId()).append("a", 2));

MongoSpark.save(jsc.parallelize(docs));

// Set the schema using the ObjectId helper

StructType schema &＃61; DataTypes.createStructType(asList(

StructFields.objectId("_id", false),

DataTypes.createStructField("a", DataTypes.IntegerType, false)));

// Create a dataframe with the helper functions registered

df &＃61; MongoSpark.read(sparkSession).schema(schema).option("registerSQLHelperFunctions", "true").load();

// Query using the ObjectId string

df.filter(format("_id &＃61; ObjectId(&＃39;%s&＃39;)", objectId)).show();

}

private static JavaSparkContext createJavaSparkContext(final String[] args) {

String uri &＃61; getMongoClientURI(args);

dropDatabase(uri);

SparkConf conf &＃61; new SparkConf()

.setMaster("local")

.setAppName("MongoSparkConnectorTour")

.set("spark.app.id", "MongoSparkConnectorTour")

.set("spark.mongodb.input.uri", uri)

.set("spark.mongodb.output.uri", uri);

return new JavaSparkContext(conf);

}

private static String getMongoClientURI(final String[] args) {

String uri;

if (args.length &＃61;&＃61; 0) {

uri &＃61; "mongodb://localhost/test.coll"; // default

} else {

uri &＃61; args[0];

}

return uri;

}

private static void dropDatabase(final String connectionString) {

MongoClientURI uri &＃61; new MongoClientURI(connectionString);

new MongoClient(uri).dropDatabase(uri.getDatabase());

}

}

。