使用 Spark 分区

Question

我是 Spark 新手，有一些关于 Spark RDD 操作和创建的问题：

val rdd1 =  sc.parallelize(List("yellow","red","blue","cyan","black"),3)
val mapped =   rdd1.mapPartitionsWithIndex{(index, iterator) => {
                                println("Called in Partition -> " + index)
                                val myList = iterator.toList
                                myList.map(x => x + " -> " + index).iterator
                               }
                            }

上面代码末尾的.iterator有什么用？它将列表转换为迭代器吗？列表本身不就是一个迭代器吗，我们到底为什么需要这个操作呢？
另外，为什么这比普通的 map() 函数快？由于分区中的每个元素都再次使用map(x => x + " -> " + index) 函数，这难道不是另一种逐个元素工作的方式吗？

另一件事，我想通过一次读取 4 行文件来创建 RDD。我在 Scala 中有以下代码：

val hconf = new org.apache.hadoop.conf.Configuration
hconf.set("mapreduce.input.lineinputformat.linespermap","4")
val line = sc.newAPIHadoopFile(inputFile,classOf[NLineInputFormat],classOf[LongWritable],classOf[Text],hconf).map(_._2.toString)
line.take(1).foreach(println)

但输出仍然只打印一行。由于我已将hconf 设置为读取 4 行，RDD 中的每个元素不应该接收 4 行 inptFile 吗？那么它不应该输出四行吗？

Answer 1

为什么要使用 .iterator？

mapPartitions 的函数参数是：

f: Iterator[T] => Iterator[U]

您粘贴的代码将每个迭代器转换为列表进行处理，并且需要在闭包结束时将其转换回迭代器以正确进行类型检查。 Spark 操作通常更喜欢流式传输数据，而不是一次将所有数据存储在内存中，并且强制分区包含 Iterators 是该模型的一部分。

关于您的“列表是一个迭代器”断言，这并不完全正确——虽然List 是Iterable，但它不是Iterator。 Iterators 的特殊之处在于它们只能使用一次，因此它们不支持许多标准的 scala Collection 操作。 Iterator 和 Iterable 之间的主要区别在于这种“一次性”模型：Iterable[T] 可以根据需要多次生成 Iterator[T]，但如果您只有 Iterator[T]，则只能查看一次。

更高效的List-free 实现

您粘贴的代码效率极低。您最终将所有数据复制到一个列表中，然后从该列表中生成一个迭代器。你可以只映射迭代器：

val rdd1 =  sc.parallelize(List("yellow","red","blue","cyan","black"),3)
val mapped =   rdd1.mapPartitionsWithIndex{(index, iterator) => {
                            println("Called in Partition -> " + index)
                            iterator.map(x => x + " -> " + index)
                           }
                        }  

每张地图的行数

我认为您可能在这里设置了错误的配置参数。请参阅this question 了解可能的解决方案。