使用Scala的Spark Appln中的flatmap可变列表

Question

我是Spark-Scala Development的新手，并试图让手弄脏，所以如果您觉得这个问题很愚蠢，请忍受我。

Sample dataset
[29430500,1104296400000,1938,F,11,2131,
MutableList([123291654450,1440129600000,100121,0,1440734400000],[234564535,2345129600000,345121,1,14567734400000])
]

如果您看到了最后一个字段，那是Array[]，我希望输出看起来像这样： -

Row 1:
    [29430500,1104296400000,1938,F,11,2131,
    123291654450,1440129600000,100121,0,1440734400000]
Row 2: 
    [29430500,1104296400000,1938,F,11,2131,
    234564535,2345129600000,345121,1,14567734400000]

我认为我必须做flatMap，但是由于某种原因，以下代码给出了此错误：

def getMasterRdd(sc: SparkContext, hiveContext: HiveContext, outputDatabase:String, jobId:String,MasterTableName:String, dataSourceType: DataSourceType, startDate:Long, endDate:Long):RDD[Row]={}
val Rdd1= ClassName.getMasterRdd(sc, hiveContext, "xyz", "test123", "xyz.abc", DataSourceType.SS, 1435723200000L, 1451538000000L)
Rdd1: holds the sample dataset
val mapRdd1= Rdd1.map(Row => Row.get(6))
val flatmapRdd1 = mapPatientRdd.flatMap(_.split(","))

当我悬停在(_.split(","))上时，我会得到一个建议，说明：

Type mismatch, expected:(Any) => TraversableOnce[NotInferedU], actual: (Any) =>Any 

Answer 1

我认为有一种更好的构造方法（也许使用元组而不是List s），但无论如何这对我有用：

scala>  val myRDD = sc.parallelize(Seq(Seq(29430500L,1104296400000L,1938L,"F",11L,2131L,Seq(Seq(123291654450L,1440129600000L,100121L,0L,1440734400000L),Seq(234564535L,2345129600000L,345121L,1L,14567734400000L)))))
myRDD: org.apache.spark.rdd.RDD[Seq[Any]] = ParallelCollectionRDD[11] at parallelize at <console>:27
scala> :pa
// Entering paste mode (ctrl-D to finish)
val myRDD2 = myRDD.flatMap(row => {
    val (beginning, end) = (row.dropRight(1), row.last)
    end.asInstanceOf[List[List[Any]]].map(beginning++_)
})
// Exiting paste mode, now interpreting.
myRDD2: org.apache.spark.rdd.RDD[Seq[Any]] = MapPartitionsRDD[10] at flatMap at <console>:29
scala> myRDD2.foreach{println}
List(29430500, 1104296400000, 1938, F, 11, 2131, 123291654450, 1440129600000, 100121, 0, 1440734400000)
List(29430500, 1104296400000, 1938, F, 11, 2131, 234564535, 2345129600000, 345121, 1, 14567734400000)

Answer 2

使用：

rdd.flatMap(row => row.getSeq[String](6).map(_.split(","))