import org.apache.hadoop.tools.util.ThrottledInputStream; //导入依赖的package包/类
private long copyBytes(FileStatus sourceFileStatus, OutputStream outStream,
                       int bufferSize, Mapper.Context context)
    throws IOException {
  Path source = sourceFileStatus.getPath();
  byte buf[] = new byte[bufferSize];
  ThrottledInputStream inStream = null;
  long totalBytesRead = 0;

  try {
    inStream = getInputStream(source, context.getConfiguration());
    int bytesRead = readBytes(inStream, buf);
    while (bytesRead >= 0) {
      totalBytesRead += bytesRead;
      outStream.write(buf, 0, bytesRead);
      updateContextStatus(totalBytesRead, context, sourceFileStatus);
      bytesRead = inStream.read(buf);
    }
  } finally {
    IOUtils.cleanup(LOG, outStream, inStream);
  }

  return totalBytesRead;
}
 

import org.apache.hadoop.tools.util.ThrottledInputStream; //导入依赖的package包/类
@VisibleForTesting
long copyBytes(FileStatus sourceFileStatus, OutputStream outStream,
                       int bufferSize, Mapper.Context context)
    throws IOException {
  Path source = sourceFileStatus.getPath();
  byte buf[] = new byte[bufferSize];
  ThrottledInputStream inStream = null;
  long totalBytesRead = 0;

  try {
    inStream = getInputStream(source, context.getConfiguration());
    int bytesRead = readBytes(inStream, buf);
    while (bytesRead >= 0) {
      totalBytesRead += bytesRead;
      outStream.write(buf, 0, bytesRead);
      updateContextStatus(totalBytesRead, context, sourceFileStatus);
      bytesRead = inStream.read(buf);
    }
    outStream.close();
    outStream = null;
  } finally {
    IOUtils.cleanup(LOG, outStream, inStream);
  }

  return totalBytesRead;
}
 

import org.apache.hadoop.tools.util.ThrottledInputStream; //导入依赖的package包/类
@VisibleForTesting
long copyBytes(FileStatus sourceFileStatus, long sourceOffset,
    OutputStream outStream, int bufferSize, Mapper.Context context)
    throws IOException {
  Path source = sourceFileStatus.getPath();
  byte buf[] = new byte[bufferSize];
  ThrottledInputStream inStream = null;
  long totalBytesRead = 0;

  try {
    inStream = getInputStream(source, context.getConfiguration());
    int bytesRead = readBytes(inStream, buf, sourceOffset);
    while (bytesRead >= 0) {
      totalBytesRead += bytesRead;
      if (action == FileAction.APPEND) {
        sourceOffset += bytesRead;
      }
      outStream.write(buf, 0, bytesRead);
      updateContextStatus(totalBytesRead, context, sourceFileStatus);
      bytesRead = readBytes(inStream, buf, sourceOffset);
    }
    outStream.close();
    outStream = null;
  } finally {
    IOUtils.cleanup(LOG, outStream, inStream);
  }
  return totalBytesRead;
}
 

import org.apache.hadoop.tools.util.ThrottledInputStream; //导入依赖的package包/类
private static int readBytes(ThrottledInputStream inStream, byte buf[],
    long position) throws IOException {
  try {
    if (position == 0) {
      return inStream.read(buf);
    } else {
      return inStream.read(position, buf, 0, buf.length);
    }
  } catch (IOException e) {
    throw new CopyReadException(e);
  }
}
 

import org.apache.hadoop.tools.util.ThrottledInputStream; //导入依赖的package包/类
private static ThrottledInputStream getInputStream(Path path,
    Configuration conf) throws IOException {
  try {
    FileSystem fs = path.getFileSystem(conf);
    long bandwidthMB = conf.getInt(DistCpConstants.CONF_LABEL_BANDWIDTH_MB,
            DistCpConstants.DEFAULT_BANDWIDTH_MB);
    FSDataInputStream in = fs.open(path);
    return new ThrottledInputStream(in, bandwidthMB * 1024 * 1024);
  }
  catch (IOException e) {
    throw new CopyReadException(e);
  }
}
 

import org.apache.hadoop.tools.util.ThrottledInputStream; //导入依赖的package包/类
private static ThrottledInputStream getInputStream(Path path,
    Configuration conf) throws IOException {
  try {
    FileSystem fs = path.getFileSystem(conf);
    float bandwidthMB = conf.getFloat(DistCpConstants.CONF_LABEL_BANDWIDTH_MB,
            DistCpConstants.DEFAULT_BANDWIDTH_MB);
    FSDataInputStream in = fs.open(path);
    return new ThrottledInputStream(in, bandwidthMB * 1024 * 1024);
  }
  catch (IOException e) {
    throw new CopyReadException(e);
  }
}
 

import org.apache.hadoop.tools.util.ThrottledInputStream; //导入依赖的package包/类
private static ThrottledInputStream getInputStream(Path path, Configuration conf)
        throws IOException {
  try {
    FileSystem fs = path.getFileSystem(conf);
    long bandwidthMB = conf.getInt(DistCpConstants.CONF_LABEL_BANDWIDTH_MB,
            DistCpConstants.DEFAULT_BANDWIDTH_MB);
    return new ThrottledInputStream(new BufferedInputStream(fs.open(path)),
            bandwidthMB * 1024 * 1024);
  }
  catch (IOException e) {
    throw new CopyReadException(e);
  }
}