我正在尝试使用 MongoDB Hadoop-Connector 将一些 JSON 从 Spark (Scala) 保存到 MongoDB。我遇到的问题是这个 API 似乎总是将您的数据保存为“{_id: ..., value: {your JSON document}}”。

在下面的代码示例中，我的文档是这样保存的:

{
    "_id" : ObjectId("55e80cfea9fbee30aa703261"),
    "value" : {
        "_id" : "55e6c65da9fbee285f2f9175",
        "year" : 2014,
        "month" : 5,
        "day" : 6,
        "hour" : 18,
        "user_id" : 246
    }
}

有什么方法可以说服 MongoDB Hadoop 连接器以您指定的结构编写 JSON/BSON，而不是将其嵌套在这些 _id/value 字段下？

这是我的 Scala Spark 代码:

  val jsonstr = List("""{
        "_id" : "55e6c65da9fbee285f2f9175",
        "year" : 2014,
        "month" : 5,
        "day" : 6,
        "hour" : 18,
        "user_id" : 246}""")

  val conf = new SparkConf().setAppName("Mongo Dummy").setMaster("local[*]")
  val sc = new SparkContext(conf)

  // DB params
  val host = "127.0.0.1"
  val port = "27017"
  val database = "dummy"
  val collection = "fubar"

  // input is collection we want to read (not doing so here)
  val mongo_input = s"mongodb://$host/$database.$collection"
  // output is collection we want to write
  val mongo_output = s"mongodb://$host/$database.$collection"

  // Set up extra config for Hadoop connector
  val hadoopConfig = new Configuration()

  //hadoopConfig.set("mongo.input.uri", mongo_input)
  hadoopConfig.set("mongo.output.uri", mongo_output)

  // convert JSON to RDD
  val rdd = sc.parallelize(jsonstr)

  // write JSON data to DB
  val saveRDD = rdd.map { json =>
    (null, Document.parse(json))
  }

  saveRDD.saveAsNewAPIHadoopFile("file:///bogus",
    classOf[Object],
    classOf[BSONObject],
    classOf[MongoOutputFormat[Object, BSONObject]],
    hadoopConfig)
  // Finished  
  sc.stop

这是我的 SBT:

name := "my-mongo-test"

version := "1.0"

scalaVersion := "2.10.4"

// Spark needs to appear in SBT BEFORE Mongodb connector!
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.4.0"

// MongoDB-Hadoop connector
libraryDependencies += "org.mongodb.mongo-hadoop" % "mongo-hadoop-core" % "1.4.0"

老实说，我对从 Spark 中保存 JSON --> BSON --> MongoDB 似乎有多难感到困惑。因此，欢迎就如何更灵活地保存我的 JSON 数据提出任何建议。

最佳答案

好吧，我刚刚找到了解决方案。结果是 MongoOutputFormat 使用的 MongoRecordWriter 插入了任何不继承自 BSONWritable 或 的值>MongoOutput 或 value 字段下的 BSONObject。

因此，最简单的解决方案是创建包含 BSONObject 作为值的 RDD，而不是 Document。

我在 Java 中尝试过这个解决方案，但我确信它在 Scala 中也能正常工作。这是一个示例代码:

JavaPairRDD<Object, BSONObject> bsons = values.mapToPair(lineValues -> {
    BSONObject doc = new BasicBSONObject();
    doc.put("field1", lineValues.get(0));
    doc.put("field2", lineValues.get(1));
    return new Tuple2<Object, BSONObject>(UUID.randomUUID().toString(), doc);
});

Configuration outputConfig = new Configuration();
outputConfig.set("mongo.output.uri",
                 "mongodb://localhost:27017/my_db.lines");
bsons.saveAsNewAPIHadoopFile("file:///this-is-completely-unused"
        , Object.class
        , BSONObject.class
        , MongoOutputFormat.class
        , outputConfig);

关于json - Apache Spark 和 MongoDB Hadoop-Connector 创建的 BSON 结构，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/32371620/