从 LIST[Row] 在 scala 中制作 RDD（在 spark 中）答案

【问题标题】：Make RDD from LIST[Row] In scala(in spark)从 LIST[Row] 在 scala 中制作 RDD（在 spark 中）
【发布时间】：2016-06-20 17:29:01
【问题描述】：

我正在使用 scala 和 spark 编写一些代码，并希望从 RDD 或 LIST[Row] 中创建 CSV 文件。

我想并行处理“ListRDD”数据，所以我的输出数据将不止一个文件。

val conf = new SparkConf().setAppName("Csv Application").setMaster("local[2]")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val logFile = "data.csv "
val rawdf = sqlContext.read.format("com.databricks.spark.csv")....
val rowRDD = rawdf.map { row =>
  Row(
    row.getAs( myMap.ID).toString,
    row.getAs( myMap.Dept)
    .....
  )
}
 val df = sqlContext.createDataFrame(rowRDD, mySchema)
val MapRDD = df.map { x => (x.getAs[String](myMap.ID), List(x)) }
val ListRDD = MapRDD.reduceByKey { (a: List[Row], b: List[Row]) => List(a, b).flatten }

myClass.myFunction( ListRDD)

在我的班级中..

def myFunction(ListRDD: RDD[(String, List[Row])]) = {
    var rows: RDD[Row]
    ListRDD.foreach( row => { 
        rows.add? gather? ( make(row._2)) // make( row._2) will return List[Row]
    })
    rows.saveAsFile(" path") // it's my final goal
  }

def make( list: List[Row]) : List[Row] = {
    data processing from List[Row]
}

我试图通过 sc.parallelize(list) 从 List 制作 RDD 数据，但不知何故没有任何效果。从 make 函数生成 RDD 类型数据的任何想法。

【问题讨论】：

这个问题很难理解：预期的输出是什么？例如，如果myFunction 收到带有两条记录的 RDD，每条记录在列表中包含两行 - 你希望输出是什么？这4行？如果是这样 - 使用reduceByKey 的目的是什么 - 它是否用于执行某种排序？尝试澄清输入和预期输出以及究竟出了什么问题（“不知何故没有任何作用”并不能说明太多）。

标签： scala csv apache-spark rdd

【解决方案1】：

如果你想从 List[Row] 中创建一个 RDD，这里有一个方法

//Assuming list is your List[Row]
val newRDD: RDD[Object] = sc.makeRDD(list.toArray());

【讨论】：