java - MongoSpark 保存重复键错误 E11000

Question

我有一个将 RDD 数据写入 MongoDB 的 Spark 应用程序，但我遇到了 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，对吗？

Answer 1

Spark 连接器不知道如何更新插入 RDD<T>其中 T可以是任何类型——如何获取 id 值？

但是，Datasets/DataFrames 具有模式信息，指示哪个字段是 _id字段并且可以自动用于更新插入。这是在 SPARK-66 中完成的. Datasets/DataFrames 的另一个好处是它们更高效，应该可以提高您的 Spark 作业的性能。

如果您必须使用 RDD，那么您可以通过编程方式访问 MongoDB 集合并通过 MongoConnector 创建更新插入操作类。