Spark RDD - 它们是如何工作的

Question

我有一个小型Scala程序,可以在单个节点上运行.但是,我正在扩展它,因此它在多个节点上运行.这是我的第一次尝试.我只是想了解RDD如何在Spark中工作,所以这个问题是基于理论的,可能不是100%正确.

假设我创建了一个RDD: val rdd = sc.textFile(file)

现在,一旦我这样做了,这是否意味着文件at file现在在节点之间进行分区(假设所有节点都可以访问文件路径)？

其次,我想计算RDD中的对象数量(足够简单),但是,我需要在需要应用于RDD中的对象的计算中使用该数字 - 伪代码示例:

rdd.map(x => x / rdd.size)

假设有100个对象rdd,并且说有10个节点,因此每个节点有10个对象的计数(假设这是RDD概念的工作方式),现在当我调用该方法时,每个节点将使用rdd.sizeas 执行计算10还是100？因为,总的来说,RDD是大小100但在每个节点上本地只是10.我是否需要在进行计算之前制作广播变量？这个问题与下面的问题有关.

最后,如果我转换到RDD,例如rdd.map(_.split("-")),然后我想要新size的RDD,我是否需要在RDD上执行操作,例如count(),所以所有信息都被发送回驱动程序节点？

Answer 1

val rdd = sc.textFile(file)

这是否意味着文件现在跨节点分区？

该文件保留在任何地方.结果的元素是RDD[String]文件的行.RDD被分区以匹配底层文件系统的自然分区.分区数不依赖于您拥有的节点数.

重要的是要理解,当执行此行时,它不会读取文件.RDD是一个懒惰的对象,只有在必要时才会执行某些操作.这很好,因为它避免了不必要的内存使用.

例如,如果你写val errors = rdd.filter(line => line.startsWith("error")),仍然没有任何反应.如果你val errorCount = errors.count现在写,那么你需要执行你的操作序列,因为结果count是一个整数.然后,每个工作者核心(执行程序线程)并行执行的操作是读取文件(或文件片段),遍历其行,并计算以"错误"开头的行.除了缓冲和GC之外,每个核心只有一行将一次存储在内存中.这使得可以在不使用大量内存的情况下处理非常大的数据.

我想计算RDD中的对象数量,但是,我需要在计算中使用该数字,该计算需要应用于RDD中的对象 - 伪代码示例:

rdd.map(x => x / rdd.size)

没有rdd.size方法.有rdd.count,它计算RDD中的元素数量.rdd.map(x => x / rdd.count)不管用.代码将尝试将rdd变量发送给所有工作人员,并将失败NotSerializableException.你能做的是:

val count = rdd.count
val normalized = rdd.map(x => x / count)

这是有效的,因为它count是一个Int可以序列化的.

如果我转换到RDD,例如rdd.map(_.split("-")),然后我想要RDD的新大小,我是否需要在RDD上执行操作,例如count(),所以所有信息都被发送回驱动程序节点？

map不会改变元素的数量.我不知道你的"大小"是什么意思.但是,是的,您需要执行一个操作,例如count从RDD中获取任何内容.你看,在你执行一个动作之前,根本不会执行任何工作.(当您执行时count,只有每个分区计数将被发送回驱动程序,当然,不是"所有信息".)

Answer 2

通常,文件(或文件的一部分,如果它太大)被复制到集群中的N个节点(默认情况下,HDFS上的N = 3).并不打算在所有可用节点之间拆分每个文件.

但是,对于您(即客户端)使用Spark处理文件应该是透明的 - rdd.size无论有多少节点被分割和/或复制,您都不应该看到任何差异.有一些方法(至少在Hadoop中)可以找出目前可以找到哪些节点(部分)文件.但是,在简单的情况下,您很可能不需要使用此功能.

更新:描述RDD内部的文章:https://cs.stanford.edu/~matei/papers/2012/nsdi_spark.pdf