代码之家 › 专栏 › 技术社区 › MetallicPriest

如何在Scala中使用mapPartitions?

apache-spark hadoop scala java

4

MetallicPriest · 技术社区 · 11 年前

我正在尝试使用 mapPartitions 但出现以下错误。

[error]  found   : Unit
[error]  required: Iterator[?]
[error] Error occurred in an application involving default arguments.
[error]         rdd.mapPartitions(showParts)

我打电话给 映射分区 功能如下。

rdd.mapPartitions(showParts)

哪里 showParts 函数的定义如下。

def showParts(iter: Iterator[(Long, Array[String])]) = 
{ 
  while (iter.hasNext)
  {
    val cur = iter.next;
    // Do something with cur
  }
}

正确的使用方法是什么 映射分区 在这里

2 回复 | 直到 11 年前

1

3

Till Rohrmann 11 年前

问题是您传递给的UDF mapPartitions 必须具有返回类型 Iterator[U] 。当前代码不返回任何内容,因此属于类型 Unit .

如果要获取空 RDD 在执行 映射分区 则可以执行以下操作:

def showParts(iter: Iterator[(Long, Array[String])]) = 
{ 
  while (iter.hasNext)
  {
    val cur = iter.next;
    // Do something with cur
  }

  // return Iterator[U]
  Iterator.empty 
}

2

1

Peter Neyens 11 年前

您需要返回 Iterator 从您的 showParts 作用

def onlyEven(numbers: Iterator[Int]) : Iterator[Int] = 
  numbers.filter(_ % 2 == 0)

def partitionSize(numbers: Iterator[Int]) : Iterator[Int] = 
  Iterator.single(numbers.length)

val rdd = sc.parallelize(0 to 10)
rdd.mapPartitions(onlyEven).collect()
// Array[Int] = Array(0, 2, 4, 6, 8, 10)

rdd.mapPartitions(size).collect()
// Array[Int] = Array(2, 3, 3, 3)