使用 sparksql 在 mongodb 中加载 JSON 数据

Question

我有一个 JSON 文件，我想使用 Spark SQL 将其加载到 MongoDB 中。我有办法将单个元素加载到下面的集合中

val mongoClient = MongoClient(127.0.0.1, 27017)
val collection = mongoClient(dbname)(collection_name)
for (a <- 1 to 10) {
  collection.insert {
    MongoDBObject("id" -> a.toString,
      "age" -> (10 + a),
      "description" -> s"description $a",
      "enrolled" -> (a % 2 == 0),
      "name" -> s"Name $a")
  }

既然 MongoDB 已经以 JSON 格式存储数据，有没有办法直接加载我的 JSON 文件？

Answer 1

当然，MongoDB 的底层数据格式是BSON，但请记住，我们可以使用mongoexport 将MongoDB 文档导出为json 格式。也许这与您的情况无关，我的意思是我们实际上可以使用SparkSQL 加载 json 文件，以下是我尝试完成的方法。

这里我们需要的一个重要库是Spark library for easy MongoDB access。

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import com.stratio.datasource.mongodb._
import com.stratio.datasource.mongodb.MongodbConfig._

object DFToMongoDB extends App {

  lazy val sc = new SparkContext(new SparkConf()
    .setAppName("mongodb")
    .setMaster("local[4]"))

  val sqlContext = new SQLContext(sc)

  /*
    sample data in xxx.json  

    {
      "_id":"xxxx","workHome":false,"commute":true,
      "tel":false,"weekend":true,"age":100.0
    }
  */
  val dataFrame = sqlContext.read.json("xxx.json")

  def forCollection(collectionName: String) = {
    MongodbConfigBuilder(
      Map(
        Host -> List("127.0.0.1"), Database -> "xxx", 
        Collection -> collectionName,
        SamplingRatio -> 1.0, 
        WriteConcern -> com.mongodb.casbah.WriteConcern.Acknowledged,
        SplitSize -> 8, SplitKey -> "_id"
      )
    ).build
  }

  // dataFrame will be converted to MongodbDataFrame here 
  dataFrame.saveToMongodb(forCollection("xxx"))
}

此外，我尝试的情况是加载之前MongoDB中不存在集合，之后MongoDB将使用json文件中的记录创建集合。至于其他情况，你可以试试看。

不管怎样，希望对你有帮助。