java – MongoSpark保存重复的密钥错误E11000

2023年1月30日 391次阅读

我有一个Spark应用程序将RDD数据写入MongoDB,我得到一个MongoBulkWriteException.以前,我使用MongoDB标准驱动程序中的bulkWrite()方法,但我已经开始使用MongoSpark驱动程序中的write()方法.

在此之前,我正在使用Apache Spark 1.6.0和MongoDB 3.2.11.

这个异常跟踪：

com.mongodb.MongoBulkWriteException: Bulk write operation error on server
 10.1.101.146:27017. Write errors: [BulkWriteError{index=0, code=11000, 
message='E11000 duplicate key error collection: collection-test 
index: _id_ dup key: { : "636253651-2017-03-07" }', details={ }}]

产生它的代码是：

JavaRDD<Document> rddInsertRecords = rddGrouped.map(new Function<Tuple2<String, BasicRecord>, Document>() {
private static final long serialVersionUID = 1L;
    @Override
    public Document call(Tuple2<String, BasicRecord> tuple2) throws Exception {
          Document json = tuple2._2.toBSONDocument();
          return json;
      }
});
MongoSpark.save(rddInsertRecords, WriteConfig.create(sc.getConf()));

我有一个使用旧代码的替代解决方案,但我想使用MongoSpark编写.

我在MongoDB的JIRA(https://jira.mongodb.org/browse/SERVER-14322)中看到过这个问题,但我不确定如何绕过这个问题.

更新：我忘了提到第一次没有发生故障(即mongodb没有数据,集合是空的).它在第二次运行作业时失败.技术上司机应该做一个upsert,对吗？

最佳答案 Spark Connector不知道如何upped RDD< T>其中T可以是任何类型 – 它如何获得id值？

但是,Datasets / DataFrames具有架构信息,指示哪个字段是_id字段,并且可以自动用于upsert.这是在SPARK-66中完成的.数据集/数据框架的另一个好处是它们更有效,并且可以提高Spark作业的性能.

如果必须使用RDD,则可以通过编程方式访问MongoDB集合并通过MongoConnector类创建upsert操作.

点赞