【Spark六十五】slice、partition和inputsplit

• slice和paritition是spark中的通用概念，表示一个原始数据块；比如使用SparkContext.parallelize方法可以把一个元素集合(不论多少元素）可以分解为多个分区(partition)
• Spark的partition可以是MapReduce中的input split，input split是Hadoop中的数据块的逻辑表示方式，input split可以一个或多个HDFS的block，input split也是Hadoop MapReduce的输入输出
• Spark的task是至计算单元，由输入数据片(partition)、处理逻辑和输出数据组成，slice或者parition是数据表示方式，仅仅是task中的“输入数据”或者“输出数据”。在ResultTask中返回值是一个数组，而ShuflleMapTask返回的是一个MapStatus，它是一个分片数据的存储信息？

Hadoop inputplit

• Hadoop的一个Block默认是128M，那么对于一个记录行形式的文本，会不会造成一行记录被分到两个Block当中？
• 在把文件从Block中读取出来进行切分时，会不会造成一行记录被分成两个InputSplit，如果被分成两个InputSplit，这样一个 InputSplit里面就有一行不完整的数据，那么处理这个InputSplit的Mapper会不会得出不正确的结果？

对于上面的两个问题，首先要明确两个概念：Block和InputSplit：

 

• Block是HDFS物理存储文件的单位（默认是128M）；
• InputSplit是MapReduce对文件进行处理和运算的输入单位，只是一个逻辑概念，每个InputSplit并没有对文件实际的切割，只是记录了要处理的数据的位置（包括文件的path和hosts）和长度（由start和length决定）。因此以行记录形式的文本，可能存在一行记录被划分到不同的Block，甚至不同的DataNode上去。通过分析FileInputFormat里面的getSplits方法，可以得出，某一行记录同样也可能被划分到不同的InputSplit

看看Hadoop关于InputSplit的获取以及都去InputSplit中文本行的代码：

 

org.apache.hadoop.mapred.FileInputFormat：

 

public InputSplit[] getSplits(JobConf job, int numSplits)
    throws IOException {
    FileStatus[] files = listStatus(job);
     
    // Save the number of input files for metrics/loadgen
    job.setLong(NUM_INPUT_FILES, files.length);
    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);
 
    // generate splits
    ArrayList<FileSplit> splits = new ArrayList<FileSplit>(numSplits);
    NetworkTopology clusterMap = new NetworkTopology();
    for (FileStatus file: files) {
      Path path = file.getPath();
      FileSystem fs = path.getFileSystem(job);
      long length = file.getLen();
      BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
      if ((length != 0) && isSplitable(fs, path)) {
        long blockSize = file.getBlockSize();
        long splitSize = computeSplitSize(goalSize, minSize, blockSize);
 
        long bytesRemaining = length;
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
          String[] splitHosts = getSplitHosts(blkLocations,
              length-bytesRemaining, splitSize, clusterMap);
          splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                               splitHosts));
          bytesRemaining -= splitSize;
        }
         
        if (bytesRemaining != 0) {
          splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
                     blkLocations[blkLocations.length-1].getHosts()));
        }
      } else if (length != 0) {
        String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);
        splits.add(makeSplit(path, 0, length, splitHosts));
      } else {
        //Create empty hosts array for zero length files
        splits.add(makeSplit(path, 0, length, new String[0]));
      }
    }
    LOG.debug("Total # of splits: " + splits.size());
    return splits.toArray(new FileSplit[splits.size()]);
  }

 

 

org.apache.hadoop.mapred.TextInputFormat:

public RecordReader<LongWritable, Text> getRecordReader(
                                          InputSplit genericSplit, JobConf job,
                                          Reporter reporter)
    throws IOException {
     
    reporter.setStatus(genericSplit.toString());
    return new LineRecordReader(job, (FileSplit) genericSplit);
  }

 

 

参考：http://my.oschina.net/xiangchen/blog/99653