spark mongodb 写入操作-基于java

迷南。 2022-12-02 00:53 107阅读 0赞

## 方式一 ##

通过Dataset.format方式写入  
mode: overwrite append

Dataset<User> userDataset = ....;
    userDataset.write().format("com.mongodb.spark.sql.DefaultSource")
                    .option("spark.mongodb.output.uri", "mongodb://test:test123@127.0.0.1:27010")
                    .option("spark.mongodb.output.database", "test")
                    .option("spark.mongodb.output.collection", "user").mode("overwrite").save();

## 方式二 ##

Map<String, String> writerConfigMap = new HashMap<String, String>();
    writerConfigMap.put("uri", "mongodb://test:test123@127.0.0.1:27010");
    writerConfigMap.put("database", "test");
    writerConfigMap.put("collection", "user");
    WriteConfig writeConfig = WriteConfig.create(this.sqlContext).withOptions(writerConfigMap);

此方式官网提供了这么一个配置参数：输出存在\_id(创建在objectId("")的数据)相同的将更新写入。但是实际操作起来会报错，报\_id值存在，插入不成功。此错误解决方案一直未找到。  
**replaceDocument**  
Replace the whole document when saving Datasets that contain an \_id field. If false it will only update the fields in the document that match the fields in the Dataset.  
**Default: true**  
错误示例, 下面代码(代码示例来自[官方demo][demo])执行两次，将会看到错误输出

String objectId = "123400000000000000000000";
      List<Document> docs = asList(
               new Document("_id", new ObjectId(objectId)).append("a", 1),
               new Document("_id", new ObjectId()).append("a", 2));
       MongoSpark.save(jsc.parallelize(docs));

## 方式三 ##

下述用到了 List分组集合，com.google.guava

Dataset<User> userDataset = ....;
    userDataset.foreachPartition(iter ->{ 
          
          List<Document> documentList = new ArrayList<>();
          for(Iterator i = iter ; iter.hasNext();){ 
                Document document = new Document("");
                document.append("", "xxxx");
               documentList.add(document);
          }
          List<List<Document>> dGroupList = Lists.partition(documentList, 100);
          
    	MongoConnector.apply(writeConfig.asOptions()).withCollectionDo(writeConfig, Document.class, new Function<MongoCollection<Document>, Void>() { 
    	                @Override
    	                public Void call(MongoCollection<Document> v1) throws Exception { 
    	
    	                    for(List<Document> d : dGroupList){ 
    	                       v1.insertMany(d);
    	                    }
    	                    return null;
    	                }
    	            });
    });

[demo]: https://github.com/mongodb/mongo-spark/blob/a13f846badc14ca08666ea402248f13958702caf/examples/src/test/java/tour/JavaIntroduction.java#L180