将 RDD[Array[Row]] 转换为 RDD[Row]答案

【问题标题】：Convert RDD[Array[Row]] to RDD[Row]将 RDD[Array[Row]] 转换为 RDD[Row]
【发布时间】：2017-08-17 14:00:51
【问题描述】：

如何将 RDD[Array[Row]] 转换为 RDD[Row]？

详情：

我有一些用例，其中我的解析函数为某些数据返回类型 Array[Row]，为某些数据返回 Row。我如何将这两个都转换为 RDD[Row] 以供进一步使用？

代码示例

private def getRows(rdd: RDD[String], parser: Parser): RDD[Row] = {

var processedLines = rdd.map { line =>      
  map(p => parser.processBeacon(line) }

val rddOfRowsList = processedLines.map { x =>
  x match {
    case Right(obj) => obj.map { p =>          
      MyRow.getValue(p)
    }//I can use flatmap here
    case Left(obj) =>          
      MyRow.getValue(obj)
  }//Cant use flatmap here
}

 // Here I have to convert rddOfRowsList to RDD[Row]
 //?????
 val rowsRdd =?????
 //


rowsRdd

}

    def processLine(logMap: Map[String, String]):Either[Map[String, Object], Array[Map[String, Object]]] =  
         {
           //process
         }

【问题讨论】：

你能添加一些你正在做什么的代码以及一些示例输入和想要的输出吗？
我遇到了类似的问题。在这里回答stackoverflow.com/q/55702513/6766464

标签： apache-spark apache-spark-sql rdd

【解决方案1】：

使用flatMap；

rdd.flatMap(identity)

【讨论】：

在我得到一行的第二种情况下，平面图将不起作用

【解决方案2】：

你可以使用 flatmap 获取新的 rdd，然后使用 union 组合它们。

【讨论】：

【解决方案3】：

使用 flatMap 扁平化 RDD 的内容

【讨论】：

虽然这可能是解决问题的宝贵提示，但一个好的答案也可以证明解决方案。请edit 提供示例代码来说明您的意思。或者，请考虑将其写为评论。