textFile构建RDD的分区及compute计算策略

1,textFile

A),第一点，就是输入格式，key，value类型及并行度的意义。

def textFile(
    path: String,
 minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
  assertNotStopped()
 //输入文件的格式TextInputFormat，key的类型LongWritable ，value的类型Text
  //最小分区数defaultMinPartitions
 hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
 minPartitions).map(pair => pair._2.toString).setName(path)
}

并行度

conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))

真正意义是啥？实际是决定我们goalSize的值。并不决定我们的分区数。

B),hadoopRDD的getPartition方法。

主要是获取分片的过程通过调用FileInputFormat.getSplits方法来实现分片。主要有一下几个步骤:

1) ,获取所有 FileStatus

FileStatus[] files = listStatus(job);

ListStatus方法里面：

1，判断是否需要递归

boolean recursive = job.getBoolean(INPUT_DIR_RECURSIVE, false);

2，接着是创建路径过滤器，筛选掉一些我们不需要的文件,入以_，.开头的

List<PathFilter> filters = new ArrayList<PathFilter>();
filters.add(hiddenFileFilter);
PathFilter jobFilter = getInputPathFilter(job);
if (jobFilter != null) {
  filters.add(jobFilter);
}
PathFilter inputFilter = new MultiPathFilter(filters);

3，根据mapreduce.input.fileinputformat.list-status.num-threads决定是并发还是单线程

FileStatus[] result;
int numThreads = job
    .getInt(
        org.apache.hadoop.mapreduce.lib.input.FileInputFormat.LIST_STATUS_NUM_THREADS,
 org.apache.hadoop.mapreduce.lib.input.FileInputFormat.DEFAULT_LIST_STATUS_NUM_THREADS);

Stopwatch sw = new Stopwatch().start();
if (numThreads == 1) {
  List<FileStatus> locatedFiles = singleThreadedListStatus(job, dirs, inputFilter, recursive); 
 result = locatedFiles.toArray(new FileStatus[locatedFiles.size()]);
} else {
  Iterable<FileStatus> locatedFiles = null;
  try {
 
    LocatedFileStatusFetcher locatedFileStatusFetcher = new LocatedFileStatusFetcher(
        job, dirs, recursive, inputFilter, false);
 locatedFiles = locatedFileStatusFetcher.getFileStatuses();
 } catch (InterruptedException e) {
 throw new IOException("Interrupted while getting file statuses");
 }
  result = Iterables.toArray(locatedFiles, FileStatus.class);
}

2) ,获取目标分片goalsize和最小minsize

long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.
  FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);

3) ,判断文件是否支持切分，不压缩或者压缩方式为BZip2Codec支持切分

protected boolean isSplitable(FileSystem fs, Path file) {
 final CompressionCodec codec = compressionCodecs.getCodec(file);
  if (null == codec) {
 return true;
 }
 return codec instanceof SplittableCompressionCodec;
}

支持切分就进行切分分片，切分分片大小为

Math.max(minSize, Math.min(maxSize, blockSize));

不支持切分的话就直接返回一个文件一个分片

最终，用InputSplit构建HadoopPartition

C),接着进入compute方法

重点掌握根据指定分片获取reader

reader = inputFormat.getRecordReader(split.inputSplit.value, jobConf, Reporter.NULL)

实际上是在TextInputFormat构建了

 new LineRecordReader(job, (FileSplit) genericSplit,
 recordDelimiterBytes);

还有就是识别不同系统的过程，比如hdfs ,本地file，tachyon。

final FileSystem fs = file.getFileSystem(job);

里面会根据uri获取scheme，然后构建为"fs." + scheme + ".impl" 通过反射的到相应的对象。

clazz = (Class<? extends FileSystem>) conf.getClass("fs." + scheme + ".impl", null);

类加载器为Configuration对象里面初始化的

private ClassLoader classLoader;
{
 classLoader = Thread.currentThread().getContextClassLoader();
  if (classLoader == null) {
 classLoader = Configuration.class.getClassLoader();
 }
}

而此，configuration对象是在compute方法中通过jobConf = getJobConf()获得的实际是

从Driver端发送过来的。

val conf: Configuration = broadcastedConf.value.value

由此可以得到结论是tachyon使用是依赖，必须方法系统类加载器的Classpath中去