flatmap使用scala在spark appln中Map可变列表

r7knjye2 于 2021-05-29 发布在 Hadoop

关注(0)|答案(2)|浏览(630)

我是新的Sparkscala的发展，并试图得到手脏，所以请与我，如果你觉得这个问题愚蠢。

Sample dataset

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

如果你看到最后一个字段 Array[] 我希望输出像this:-

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

hadoop scala apache-spark apache-spark-sql

来源：https://stackoverflow.com/questions/41027618/flatmap-a-mutable-list-in-spark-appln-using-scala

2条答案

按热度按时间

von4xj4u1#

用途：

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

赞(0）回复(0）举报 2021-05-29

92dk7w1h2#

我认为有一个更好的方法来构造它（也许使用元组而不是 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)

赞(0）回复(0）举报 2021-05-29

我来回答

flatmap使用scala在spark appln中Map可变列表

2条答案

相关问题

热门标签

最新问答