使用 Apache Spark 将 MongoDB 数据保存为 parquet 文件格式答案

【问题标题】：Save MongoDB data to parquet file format using Apache Spark使用 Apache Spark 将 MongoDB 数据保存为 parquet 文件格式
【发布时间】：2015-10-28 13:56:35
【问题描述】：

我是 Apache spark 以及 Scala 编程语言的新手。

我想要实现的是从我的本地 mongoDB 数据库中提取数据，然后使用带有 hadoop-connector 的 Apache Spark 将其保存在 parquet format 中

到目前为止，这是我的代码：

package com.examples 
import org.apache.spark.{SparkContext, SparkConf} 
import org.apache.spark.rdd.RDD 
import org.apache.hadoop.conf.Configuration 
import org.bson.BSONObject 
import com.mongodb.hadoop.{MongoInputFormat, BSONFileInputFormat} 
import org.apache.spark.sql 
import org.apache.spark.sql.SQLContext 

object DataMigrator { 

    def main(args: Array[String])
    { 
        val conf = new SparkConf().setAppName("Migration    App").setMaster("local") 
        val sc = new SparkContext(conf) 
        val sqlContext = new SQLContext(sc) 

        // Import statement to implicitly convert an RDD to a DataFrame 
        import sqlContext.implicits._ 

        val mongoConfig = new Configuration() 
        mongoConfig.set("mongo.input.uri",   "mongodb://localhost:27017/mongosails4.case") 

        val mongoRDD = sc.newAPIHadoopRDD(mongoConfig, classOf[MongoInputFormat], classOf[Object], classOf[BSONObject]);     

        val count = countsRDD.count()

        // the count value is aprox 100,000 
        println("================ PRINTING =====================") 
        println(s"ROW COUNT IS $count") 
        println("================ PRINTING =====================") 
    } 
}

问题是，为了将数据保存为 parquet 文件格式，首先需要将 mongoRDD 变量转换为 Spark DataFrame。我尝试过这样的事情：

// convert RDD to DataFrame
val myDf = mongoRDD.toDF()  // this lines throws an error
myDF.write.save("my/path/myData.parquet")

我得到的错误是： Exception in thread "main" scala.MatchError: java.lang.Object (of class scala.reflect.internal.Types.$TypeRef$$anon$6)

你们还有其他想法吗？如何将 RDD 转换为 DataFrame，以便我可以以 parquet 格式保存数据？

这是 mongoDB 集合中一个 Document 的结构：https://gist.github.com/kingtrocko/83a94238304c2d654fe4

【问题讨论】：

标签： scala hadoop apache-spark parquet mongodb-hadoop

【解决方案1】：

创建一个表示存储在 DBObject 中的数据的 Case 类。
case class Data(x: Int, s: String)

然后，将 rdd 的值映射到案例类的实例。 val dataRDD = mongoRDD.values.map { obj => Data(obj.get("x"), obj.get("s")) }

现在有了你的 RDD[Data]，你可以用 sqlContext 创建一个 DataFrame

val myDF = sqlContext.createDataFrame(dataRDD)

这应该会让你继续前进。如果需要，我可以稍后再解释。

【讨论】：