没有 MongoSpark 无法从 Spark 向 MongoDB 插入数据

Question

我在这里想要实现的是将多个文档插入到多个数据库的给定集合中。

数据库根据对象中的给定属性而有所不同，因此他们每个人都应该从一个小的缓存映射中选择（键，MongoCollection [指向其各自的数据库，并不总是相同]）他们需要和.insertOne 与文档。

我发现的问题是MongoCollectionImpl 在执行以下操作时似乎不可序列化（简化，但复制了错误）...

.map(
  MongoClient(s"mongodb://localhost:27017")
    .getDatabase("sample")
    .getCollection("sample_collection")
    .insertOne)

我知道 MongoDB 中有一个专门为 Spark 设计的连接器，但我不知道如何使用多个数据库，因为我需要几个 WriteConfig 应该有条件地应用于每个元素，所以它们不会被插入到它们不应该在的另一个数据库中。

https://docs.mongodb.com/spark-connector/master/scala/write-to-mongodb/

有没有与此相关的解决方法？

Answer 1

只需使用MongoSpark 连接器并根据条件对原始数据帧进行分区/拆分。

myDataFrame.cache() // recommended to prevent repeating the calculation

val df1 = myDataFrame.filter('myColumn < 0)
val df2 = myDataFrame.filter('myColumn >= 0 && 'myColumn < 50)
val df3 = myDataFrame.filter('myColumn >= 50 )

MongoSpark.save(df1.write.mode(SaveMode.Overwrite), WriteConfig(databaseName = ???, collectionName = ???, connectionString = Some(???))

MongoSpark.save(df2.write.mode(SaveMode.Overwrite), WriteConfig(databaseName = ???, collectionName = ???, connectionString = Some(???))

MongoSpark.save(df3.write.mode(SaveMode.Overwrite), WriteConfig(databaseName = ???, collectionName = ???, connectionString = Some(???))