首页 > 代码库 > spark textFile 困惑与解释

spark textFile 困惑与解释

2024-09-02 00:25:37   219人阅读

在编写spark测试应用时， 会用到sc.textFile(path, partition)

当配置为spark分布式集群时，当你读取本地文件作为输入时， 需要将文件存放在每台work节点上。

这时会有困惑，spark在读取文件时，是每台worker节点都把文件读入？ 然后在进行分配？ 会不会出现重复读的情况？ 文件会分为几个partition？

转自知乎：https://www.zhihu.com/question/36996853

 

作者：羊咩

一·是在执行action的时候再拷贝相应分区到多个worker节点进行并行计算吗？
不是，这种读取local file system而不是hdfs的情况，需要同一个文件存在所有的worker node上面，在读取的时候每个worker node的task会去读取本文件的一部分。打个比方，比如你有一个file，有一个spark集群(node1是master,node2,node3两个是slaves)，那么这个file需要在node2,node3上面都存在，这两个节点的task会各读一半，不然会出错。（这里其实还有一个点注意，你的spark app所运行的节点也需要有这个file，因为需要用到file进行Partition划分）。

二·具体对应哪一段源码。
1.由读取文件的方法SparkContext.textFile(path)跟踪源码知道它利用了TextInputFormat生成了一个HadoopRDD.
def textFile(      path: String,      minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {    assertNotStopped()    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],      minPartitions).map(pair => pair._2.toString)  }def hadoopFile[K, V](      path: String,      inputFormatClass: Class[_ <: InputFormat[K, V]],      keyClass: Class[K],      valueClass: Class[V],      minPartitions: Int = defaultMinPartitions): RDD[(K, V)] = withScope {    assertNotStopped()    // A Hadoop configuration can be about 10 KB, which is pretty big, so broadcast it.    val confBroadcast = broadcast(new SerializableConfiguration(hadoopConfiguration))    val setInputPathsFunc = (jobConf: JobConf) => FileInputFormat.setInputPaths(jobConf, path)    new HadoopRDD(      this,      confBroadcast,      Some(setInputPathsFunc),      inputFormatClass,      keyClass,      valueClass,      minPartitions).setName(path)  }

2.再来分析HadoopRDD，对于你的疑问来说最重要的是getPartitions方法，也就是如何划分你输入的文件成为Partitions：

override def getPartitions: Array[Partition] = {    val jobConf = getJobConf()    // add the credentials here as this can be called before SparkContext initialized    SparkHadoopUtil.get.addCredentials(jobConf)    val inputFormat = getInputFormat(jobConf)    if (inputFormat.isInstanceOf[Configurable]) {      inputFormat.asInstanceOf[Configurable].setConf(jobConf)    }    val inputSplits = inputFormat.getSplits(jobConf, minPartitions)    val array = new Array[Partition](inputSplits.size)    for (i <- 0 until inputSplits.size) {      array(i) = new HadoopPartition(id, i, inputSplits(i))    }    array  }

其中 val inputSplits = inputFormat.getSplits(jobConf, minPartitions), 是将你的输入文件划分为多个Split，一个Split对应一个Partition，因为是本地文件系统，通过"file://"前缀可以获取文件系统，这个源码我就不帖了，这里minPartitions是2（如果你没有指定的话），也就是将file划分为2部分，每个Split都有SplitLocationInfo描述该Split在哪个node上如何存储，比如FileSplit包含了（Hosts，start, len, path），就是在哪个host上面的哪个path，从哪个起点start读取len这么多数据就是这个Split的内容了。对于本地文件，他的Host直接指定的是localhost，path就是你传入的文件路径，start和len根据2份进行简单的计算即可，我就不赘述。有了这个信息我们可以构造每个Split的PreferLocation:
override def getPreferredLocations(split: Partition): Seq[String] = {    val hsplit = split.asInstanceOf[HadoopPartition].inputSplit.value    val locs: Option[Seq[String]] = HadoopRDD.SPLIT_INFO_REFLECTIONS match {      case Some(c) =>        try {          val lsplit = c.inputSplitWithLocationInfo.cast(hsplit)          val infos = c.getLocationInfo.invoke(lsplit).asInstanceOf[Array[AnyRef]]          Some(HadoopRDD.convertSplitLocationInfo(infos))        } catch {          case e: Exception =>            logDebug("Failed to use InputSplitWithLocations.", e)            None        }      case None => None    }    locs.getOrElse(hsplit.getLocations.filter(_ != "localhost"))  }

从这段代码可以看出来，对于localhost的host，是没有PreferredLocation的，这个会把对应于该partition的task追加到no_prefs的任务队列中，进行相应data locality的任务调度。

3.任务调度
val taskIdToLocations = try {      stage match {        case s: ShuffleMapStage =>          partitionsToCompute.map { id => (id, getPreferredLocs(stage.rdd, id))}.toMap        case s: ResultStage =>          val job = s.resultOfJob.get          partitionsToCompute.map { id =>            val p = job.partitions(id)            (id, getPreferredLocs(stage.rdd, p))          }.toMap      }    }
由于Spark每个partition的运算都是由一个task进行的，那么partition的preferlocation会成为task的preferLocation，这是data locality的任务调度，遵循着移动计算比移动数据更加高效的原则。
那么这样每个task都有了自己的应该允许的Location，然而对于本地文件系统，这是一个坑爹的存在，因为getPreferredLocs这个方法返回的是Nil，是空的。如果task没有PreferLocation，那么它如何被调度呢？答案在TaskSetManager里面：
if (tasks(index).preferredLocations == Nil) {      addTo(pendingTasksWithNoPrefs)    }
如何没有preferLocation的话，那么是会把这个任务追加到pendingTasksWithNoPrefs数组里面。
该数组里面的任务是以Round-Robin的方式分发到各个Executor里面的，到这里已经能说明问题了，你有一个file，根据FileInputFormat生成了两个Split，HadoopRDD据此生成了两个Partition，两个Partition需要两个Task，这两个Task会 Round-Robin 得spread到你的node2,node3上面的executor上面，这些Task要读取的Split的文件的host都是localhost，大小就是file的一半，到此，你应该可以理解为什么需要这个file在每个worker node都存在了，因为每个worker node的executor执行的task要读取的Split的Location信息是localhost，他不会到master上面读，只会在运行这个task的worker node本地读。相对应的源码就是上面的，细节留待你自己去再梳理一遍。

PS：
1.这种使用textFile方法读取本地文件系统的文件的方法，只能用于debug，不用于其他任何用途，因为他会导致file的replication数与node的个数同步增长。

2.上述描述中的分成2份这种是默认值，为了方面说明，你可以自己设置partition个数。

spark textFile 困惑与解释


联系
我们