spark读取hdfs的文件是怎么分区的，读取代码如下：

val df = sc.textFile("data/wc.txt",3)

闪闪发光的星星灰尘散景背景上的足迹

一.分析

spark读取hdfs的文件分区跟hadoop的分区完全相同,因为底层使用的就是Hadoop的TextInputFormat，考虑两内容：

1)关于文件分区数量计算：

指定的预分区数量是最小分区数量，如：代码中的参数3。

真正的分区计算： 每个分区字节数 = 文件字节数/预分区数量

如果没有整除，判断余数是否大于分区字节数 * 0.1，如果大于则会新增一个分区，剩余的放在这个分区。否则不会新加分区，把余数放到最后一个分区里。

2)分区数据如何读取：

分区读取数据是按行读取，但是会考虑文件的偏移量(offset)的设置。虽然第一个分区字节数不包含一整行，但是会读取一整行。当某个分区的偏移量全被之前的分读走了，这个分区就是空的。

注意：

1.当位移量读取了回撤换行，会把下一行的数据也会读取。

2.当读取多个文件时，会把所有文件字节加起来计算分区，但是读取的时候不会夸文件读取。

二.代码分析

1）读取文件数据时，数据是按照Hadoop文件读取的规则进行切片分区，而切片规则和数据读取的规则有些差异，具体Spark核心源码如下,FileInputFormat类中的方法：

public InputSplit[] getSplits(JobConf job, int numSplits)

throws IOException {

long totalSize = 0; // compute total size

for (FileStatus file: files) {// check we have valid files

if (file.isDirectory()) {

throw new IOException("Not a file: "+ file.getPath());

}

totalSize += file.getLen();

}

long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);

long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.

FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);

...

for (FileStatus file: files) {

...

if (isSplitable(fs, path)) {

long blockSize = file.getBlockSize();

long splitSize = computeSplitSize(goalSize, minSize, blockSize);

...

}

protected long computeSplitSize(long goalSize, long minSize,long blockSize) {

return Math.max(minSize, Math.min(goalSize, blockSize));

2) 分区数据读取的代码LineReader

public LineRecordReader(Configuration job, FileSplit split,    byte[] recordDelimiter) throws IOException {  this.maxLineLength = job.getInt(org.apache.hadoop.mapreduce.lib.input.    LineRecordReader.MAX_LINE_LENGTH, Integer.MAX_VALUE);  start = split.getStart();  end = start + split.getLength();  final Path file = split.getPath();  compressionCodecs = new CompressionCodecFactory(job);  codec = compressionCodecs.getCodec(file);  // open the file and seek to the start of the split  final FileSystem fs = file.getFileSystem(job);  fileIn = fs.open(file);  if (isCompressedInput()) {    decompressor = CodecPool.getDecompressor(codec);    if (codec instanceof SplittableCompressionCodec) {      final SplitCompressionInputStream cIn =        ((SplittableCompressionCodec)codec).createInputStream(          fileIn, decompressor, start, end,          SplittableCompressionCodec.READ_MODE.BYBLOCK);      in = new CompressedSplitLineReader(cIn, job, recordDelimiter);      start = cIn.getAdjustedStart();      end = cIn.getAdjustedEnd();      filePosition = cIn; // take pos from compressed stream    } else {      in = new SplitLineReader(codec.createInputStream(fileIn,          decompressor), job, recordDelimiter);      filePosition = fileIn;    }  } else {    fileIn.seek(start);    in = new UncompressedSplitLineReader(        fileIn, job, recordDelimiter, split.getLength());    filePosition = fileIn;  }

三案例分析

案例：读取文件重新分区，再写入到文件

代码：

val conf = new Configuration()val fs = FileSystem.getLocal(conf)fs.delete(new Path("data/wc.out2"),true)val df = sc.textFile("data/wc.txt",3)df.saveAsTextFile("data/wc.out2")

 文件内容：   d   a   y   u   e   e   t 1）计算分区： 总字节数：14；指定预分区为3. 目标分区大小： goalSize = 14/3 =4， 余2； 2/4>0.1,所以会生成4个分区，每个分区大小是4个字节。每个分区读取偏移量入下：   0，4   4，8   8，12   12, 132)分区读取内容 按行读取 文件内容的offset为： 0 1 2 3 4 5 6 7 8 9 10 11 12  第1个分区读取：0到4，读取前3行 ；注意：当位移量偏过换行符时，会把下一行的数据也会读取了  第2个分区读取：4到8，因为4，5被读去了，从来6开始读到9。（因为按行读取所以读到9）  第3个分区读取：8到12，因为8，9被读去了，从来10开始读到12。  第4个分区读取：12到13 所以为

四.总结

spark读取hdfs文件分区比较复杂，需要仔细研究研究。