Я хотел бы обновить определенную коллекцию в MongoDb через Spark в Java. Я использую Коннектор MongoDB для Hadoop для извлечения и сохранения информации из Apache Spark для MongoDb в Java.
После прочтения отличного сообщения Sampo Niskanen о получении и сохранении коллекций в MongoDb через Spark, Я застрял с обновлением коллекций.
MongoOutputFormat.java включает конструктор, принимающий String[] updateKeys, который, как я предполагаю, относится к возможному списку ключей для сравнения существующих коллекций и выполнения обновления. Однако, используя метод Spark saveAsNewApiHadoopFile()
с параметром MongoOutputFormat.class
, мне интересно, как использовать этот конструктор обновлений.
save.saveAsNewAPIHadoopFile("file:///bogus", Object.class, Object.class, MongoOutputFormat.class, config);
До этого MongoUpdateWritable.java использовался для обновления коллекций. Из примеров, которые я видел в Hadoop, это обычно устанавливается на mongo.job.output.value
, может быть, так в Spark:
save.saveAsNewAPIHadoopFile("file:///bogus", Object.class, MongoUpdateWritable.class, MongoOutputFormat.class, config);
Однако мне все еще интересно, как указать ключи обновления в MongoUpdateWritable.java
.
По общему признанию, в качестве хакерского способа я установил «_id» объекта в качестве значения ключа моего документа, чтобы при выполнении сохранения коллекция перезаписывала документы, имеющие то же значение ключа, что и _id
.
JavaPairRDD<BSONObject,?> analyticsResult; //JavaPairRdd of (mongoObject,result)
JavaPairRDD<Object, BSONObject> save = analyticsResult.mapToPair(s -> {
BSONObject o = (BSONObject) s._1;
//for all keys, set _id to key:value_
String id = "";
for (String key : o.keySet()){
id += key + ":" + (String) o.get(key) + "_";
}
o.put("_id", id);
o.put("result", s._2);
return new Tuple2<>(null, o);
});
save.saveAsNewAPIHadoopFile("file:///bogus", Object.class, Object.class, MongoOutputFormat.class, config);
Я хотел бы выполнить обновление коллекции mongodb через Spark, используя MongoOutputFormat
или MongoUpdateWritable
или Configuration
, в идеале используя метод saveAsNewAPIHadoopFile()
. Является ли это возможным? Если нет, есть ли другой способ, который не требует специальной установки _id для значений ключа, которые я хочу обновить?