final String mgohost = "127.0.0.1";
SparkSession spark = SparkSession.builder()
                .appName("spot")
                .config("spark.mongodb.output.uri", "mongodb://spark:[email protected]" + mgohost + ":27017/admin")
                .config("spark.mongodb.output.database","demo")
                .config
 
("spark.mongodb.output.collection","test")
                .getOrCreate();

• mongodb://使用者名稱:密碼@” + mgohost + “:27017/admin
• admin為mongodb系統database，通過系統庫連線database許可權認證通過
• demo指定其它database
• test指定collection

JavaSparkContext jc = new JavaSparkContext(spark.sparkContext());
        JavaRDD<String> spotInfo = jc.textFile(input);
JavaRDD<Document> javaRDD = spotInfo.map(new
 
 Function<String, Document>() {
            @Override
            public Document call(String arg0) {
                try {
                    Document document = new Document();
                    document.put("data", arg0);
                    return document;
                } catch (Exception
 
 e) {
                    // TODO: handle exception
                }
                Document document = new Document();
                document.put("error", "error");
                return document;
            }
        }).filter(new Function<Document, Boolean>() {
            @Override
            public Boolean call(Document arg0) throws Exception {
                if(arg0.containsKey("error")){
                    return false;
                };
                return true;
            }
        });
        MongoSpark.save(javaRDD);
        jc.stop();

• 遍歷檔案每一條資料返回Document儲存至mongodb資料庫