scala - Spark : Using mapPartition with Scala

Question

假设我有以下数据框:

var randomData = Seq(("a",8),("h",5),("f",3),("a",2),("b",8),("c",3)
val df = sc.parallelize(randomData,2).toDF()

我有这个函数，它将作为 mapPartition 的输入:

def trialIterator(row:Iterator[(String,Int)]): Iterator[(String,Int)] =
    row.toArray.tail.toIterator

并使用 map 分区:

df.mapPartition(trialIterator)

我收到以下错误消息:

Type mismatch, expected (Iterator[Row]) => Iterator[NotInferedR], actual: Iterator[(String,Int) => Iterator[(String,Int)]

我可以理解这是由于我的函数的输入、输出类型而发生的，但是如何解决这个问题？

Answer 1

如果您想获得强类型输入，请不要使用 Dataset[Row] ( DataFrame ) 但 Dataset[T]哪里T在这种特殊情况下是 (String, Int) .也不要转换为 Array不要盲目打电话tail不知道分区是否为空:

def trialIterator(iter: Iterator[(String, Int)]) = iter.drop(1)

randomData
  .toDS // org.apache.spark.sql.Dataset[(String, Int)]
  .mapPartitions(trialIterator _)

或者

randomData.toDF // org.apache.spark.sql.Dataset[Row] 
  .as[(String, Int)] // org.apache.spark.sql.Dataset[(String, Int)]
  .mapPartitions(trialIterator _)