spark讀取mongodb 並使用fastjson對讀取json進行處理

package spark_read;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;
import org.bson.Document;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;

public class Read_Mongo {

public static void main(String[] args) {
JavaSparkContext jsc = createJavaSparkContext(args);

JavaMongoRDD<Document> mongoRDD = MongoSpark.load(jsc);

mongoRDD.foreach(new VoidFunction<Document>() {

private static final long serialVersionUID = 1L;

public void call(Document document) throws Exception {

String data = document.toJson();

JSONObject jsonObject = JSON.parseObject(data);

JSONArray src = jsonObject.getJSONArray("_src");

JSONObject src_obj = (JSONObject) src.get(0);

System.out.println(src_obj.getString("site"));

}
});
}

/**
建立spark鏈接，並設置mongodb讀寫路徑信息
*/
private static JavaSparkContext createJavaSparkContext(final String[] args) {
String uri = getMongoClientURI(args);
//dropDatabase(uri);
SparkConf conf = new SparkConf()
.setMaster("local")
.setAppName("MongoSparkConnectorTest")
.set("spark.app.id", "MongoSparkConnectorTour")
.set("spark.mongodb.input.uri", uri)
.set("spark.mongodb.output.uri", uri);

return new JavaSparkContext(conf);
}

/**
刪除mongo已存在文件
*/
private static void dropDatabase(final String connectionString) {
MongoClientURI uri = new MongoClientURI(connectionString);
new MongoClient(uri).dropDatabase(uri.getDatabase());
}

/**
獲取mondo讀寫路徑
*/
private static String getMongoClientURI(final String[] args) {
String uri;
if (args.length == 0) {
uri = "mongodb://ip:27017/mongo庫名.表名"; // default
} else {
uri = args[0];
}
return uri;
}

}