Hadoop基础

@Test
public void mkdirTest() throws Exception {
  Configuration configuration = new Configuration();
  configuration.set("dfs.replication", "2");
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  boolean success = fileSystem.mkdirs(new Path("/java/api/input"));
  log.info("执行结果{}", success);
  fileSystem.close();
}

@Test
public void copyTest() throws Exception {
  Configuration configuration = new Configuration();
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  fileSystem.copyToLocalFile(false, new Path("/input/sanguo"),  new Path("F:\\input"));
  fileSystem.close();
}

@Test
public void ioInputTest() throws Exception {
  Configuration configuration = new Configuration();
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  InputStream inputStream = new FileInputStream(new File("F:\\input\\sanguo\\wuguo.txt"));
  OutputStream outputStream = fileSystem.create(new Path("/wuguo.txt"));
  IOUtils.copyBytes(inputStream, outputStream, configuration);
  fileSystem.close();
}

@Test
public void copyTest() throws Exception {
  Configuration configuration = new Configuration();
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  fileSystem.copyToLocalFile(false, new Path("/input/sanguo"),  new Path("F:\\input"));
  fileSystem.close();
}

@Test
public void ioDownTest() throws Exception {
  Configuration configuration = new Configuration();
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  OutputStream outputStream = new FileOutputStream(new File("F:\\input\\sanguo\\wuguo.txt"));
  InputStream inputStream = fileSystem.open(new Path("/wuguo.txt"));
  IOUtils.copyBytes(inputStream, outputStream, configuration);
  fileSystem.close();
}

@Test
public void rmTest() throws Exception {
  Configuration configuration = new Configuration();
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  boolean success = fileSystem.delete(new Path("/input/sanguo"), false);
  log.info("执行结果{}", success);
  fileSystem.close();
}

@Test
public void renameTest() throws Exception {
  Configuration configuration = new Configuration();
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  boolean success = fileSystem.rename(new Path("/input/sanguo"), new Path("/input/ThreeKing"));
  log.info("执行结果{}", success);
  fileSystem.close();
}

@Test
public void fileDetailTest() throws Exception {
  Configuration configuration = new Configuration();
  FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.1.161:9000"), configuration, "kun");
  /**
    * listFiles：只能列出文件，可控制是否递归
    * listLocatedStatus: 只能获取当前目录下的目录和文件
    * listStatusIterator: 和listLocatedStatus相似，但返回的是文件块状态信息少
    */
  RemoteIterator<LocatedFileStatus> listFiles = fileSystem.listLocatedStatus(new Path("/input/ThreeKing"));
  while (listFiles.hasNext()) {
    LocatedFileStatus fileStatus = listFiles.next();
    log.info("文件名:{}", fileStatus.getPath().getName());
    log.info("长度:{}", fileStatus.getLen());
    log.info("权限:{}", fileStatus.getPermission());
    log.info("是文件:{}", fileStatus.isFile());
    log.info("是目录:{}", fileStatus.isDirectory());
    // 块信息
    BlockLocation[] blockLocations = fileStatus.getBlockLocations();
    for (BlockLocation blockLocation : blockLocations) {
      String[] hosts = blockLocation.getHosts();
      for (String host : hosts) {
        log.info("存储节点:{}", host);
      }
    }
  }
  fileSystem.close();
}

@Slf4j
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

  private Text outKey = new Text();
  private IntWritable outValue = new IntWritable();

  // 任务开始时会被调用一次
  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    log.info("=================WordCountMapper setup=================");
  }

  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String text = value.toString();
    String[] words = text.split(" ");
    for (String word : words) {
      outKey.set(word);
      outValue.set(1);
      context.write(outKey, outValue);
    }
  }

  // 任务结束会被调用一次
  @Override
  protected void cleanup(Context context) throws IOException, InterruptedException {
    log.info("=================WordCountMapper cleanup=================");
  }
}

@Slf4j
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

  private IntWritable outValue = new IntWritable();

  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    log.info("=================WordCountReduce setup=================");
  }

  @Override
  protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    int count = 0;
    for (IntWritable value : values) {
      count += value.get();
    }
    outValue.set(count);
    context.write(key, outValue);
  }

  @Override
  protected void cleanup(Context context) throws IOException, InterruptedException {
    log.info("=================WordCountReduce cleanup=================");
  }
}

public class WordCountDriver {

  public static void main(String[] args) throws Exception {
    // 获取JOB
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.1.161:9000");

    Job job = Job.getInstance(configuration);
    // 设置jar位置
    job.setJarByClass(WordCountDriver.class);
    // 关联Mapper和Reducer
    job.setMapperClass(WordCountMapper.class);
    job.setReducerClass(WordCountReducer.class);
    // 设置Mapper阶段输出的KV类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.class);
    // 设置最后输出的KV类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    // 设置程序输出的输入路径和输出路径
    FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.1.161:9000/input"));
    FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.1.161:9000/output"));
    // 提交JOB
    job.submit();
    job.waitForCompletion(true);
  }
}

/**
 * 必须提供无参构造函数
 * 实现Writable接口，write和readFields字段顺序必须完全一致
 * 重新toString提供输出到文件的格式
 * 如果需要将自定义的bean放在key中传输还要实现Comparable接口
 * 必须有值，否则序列化报错
 */
@Data
@NoArgsConstructor
public class FlowerBean implements Writable {
  private String name;
  private long upFlow;
  private long downFlow;
  private long sumFlow;

  @Override
  public void write(DataOutput out) throws IOException {
    out.writeUTF(name);
    out.writeLong(upFlow);
    out.writeLong(downFlow);
    out.writeLong(sumFlow);
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    this.name = in.readUTF();
    this.upFlow = in.readLong();
    this.downFlow = in.readLong();
    this.sumFlow = in.readLong();
  }

  @Override
  public String toString() {
    return name + '\t' + upFlow + '\t' + downFlow + '\t' + sumFlow;
  }
}

public class FlowerMapper extends Mapper<LongWritable, Text, Text, FlowerBean> {

  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String lineText = value.toString();
    String[] values = lineText.split("\t");
    FlowerBean flowerBean = new FlowerBean();
    flowerBean.setName(values[1]);
    flowerBean.setUpFlow(Long.parseLong(values[values.length-3]));
    flowerBean.setDownFlow(Long.parseLong(values[values.length-2]));
    context.write(new Text(values[1]), flowerBean);
  }
}

public class FlowerReducer extends Reducer<Text, FlowerBean, Text, FlowerBean> {

  @Override
  protected void reduce(Text key, Iterable<FlowerBean> values, Context context) throws IOException, InterruptedException {

    FlowerBean flowerBean = new FlowerBean();
    long downFlow = 0;
    long upFlow = 0;
    for (FlowerBean value : values) {
      downFlow += value.getDownFlow();
      upFlow += value.getUpFlow();
    }
    long sumFlow = upFlow + downFlow;
    flowerBean.setUpFlow(upFlow);
    flowerBean.setDownFlow(downFlow);
    flowerBean.setSumFlow(sumFlow);
    flowerBean.setName(key.toString());
    context.write(key, flowerBean);
  }
}

public class FlowerDriver {

  public static void main(String[] args) throws Exception {
    // 获取JOB
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.1.161:9000");
    Job job = Job.getInstance(configuration);
    // 关联Mapper和Reducer
    job.setMapperClass(FlowerMapper.class);
    job.setReducerClass(FlowerReducer.class);
    // 设置Mapper阶段输出的KV类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(FlowerBean.class);
    // 设置最后输出的KV类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(FlowerBean.class);
    // 设置程序输出的输入路径和输出路径
    FileInputFormat.setInputPaths(job, new Path("/input/flower"));
    FileOutputFormat.setOutputPath(job, new Path("/output"));
    // 提交JOB
    job.submit();
    job.waitForCompletion(true);
  }
}

①如果最大切片大小设置为4M，文件剩余大小大于4M但小于8M则将最后两端进行平均切分
file_1.txt 10M  ->  file_1.txt.split1 4M；file_1.txt.split2 3M；file_1.txt.split3 3M
file_2.txt 2M   ->  file_2.txt.split1 2M
file_3.txt 6M   ->  file_3.txt.split1 3M；file_3.txt.split2 3M 
②进行组合，两个切片相加小于4M继续加，如果大于等于4M则停止且为一个切片
切片1：file_1.txt.split1 4M；
切片2：file_1.txt.split2 3M；file_3.txt.split1 3M；
切片3：file_2.txt.split1 2M；file_3.txt.split1 3M；
切片4：file_3.txt.split2 3M 

job.setInputFormatClass(CombineTextInputFormat.class);
CombineTextInputFormat.setMaxInputSplitSize(job, 128 * 1024 * 1024);// 128m

// 设置切割符，必须再得到job实例之前
configuration.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");
Job job = Job.getInstance(configuration);
job.setInputFormatClass(KeyValueTextInputFormat.class);

NLineInputFormat.setNumLinesPerSplit(job, 3);
job.setInputFormatClass(NLineInputFormat.class);  

public class CustomArchiveInputFormat extends FileInputFormat<Text, BytesWritable> {

  @Override
  public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
    CustomArchiveRecordReader customArchiveRecordReader = new CustomArchiveRecordReader();
    customArchiveRecordReader.initialize(split, context);
    return customArchiveRecordReader;
  }

  /* 文件是否可以在逻辑上进行切分 */
  @Override
  protected boolean isSplitable(JobContext context, Path filename) {
    return false;
  }

  /* 调用父方法，返回的列表包含文件夹，再将父方法返回的数据再次过滤 */
  @Override
  protected List<FileStatus> listStatus(JobContext job) throws IOException {
    List<FileStatus> statusList = super.listStatus(job);
    return statusList.stream().filter(FileStatus::isFile).collect(Collectors.toList());
  }
}

@Slf4j
public class CustomArchiveRecordReader extends RecordReader<Text, BytesWritable> {

  private FileSplit split;
  private TaskAttemptContext context;
  private BytesWritable currentValue = new BytesWritable();
  private Text currentKey = new Text();
  // 每次mapper都会创建一个新的，所以初始化都为true
  private boolean isBegin = true;

  @Override
  public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
    this.split = (FileSplit) split;
    this.context = context;
  }

  @Override
  public boolean nextKeyValue() throws IOException, InterruptedException {
    log.info("File【{}】开始处理", split.getPath().getName());
    if (isBegin) {
      byte[] contents = new byte[(int)split.getLength()];
      Path path = split.getPath();
      FileSystem fileSystem = path.getFileSystem(context.getConfiguration());
      FSDataInputStream inputStream = fileSystem.open(path);
      currentKey.set(path.getName());
      IOUtils.readFully(inputStream, contents, 0, contents.length);
      currentValue.set(contents, 0, contents.length);
      isBegin = false;
      return true;
    }
    return false;
  }

  @Override
  public Text getCurrentKey() throws IOException, InterruptedException {
    return currentKey;
  }

  @Override
  public BytesWritable getCurrentValue() throws IOException, InterruptedException {
    return currentValue;
  }

  @Override
  public float getProgress() throws IOException, InterruptedException {
    return 0;
  }

  @Override
  public void close() throws IOException {
    log.info("File【{}】已完成", split.getPath().getName());
  }
}

@Slf4j
public class CustomArchiveMapper extends Mapper<Text, BytesWritable, Text, BytesWritable> {

  @Override
  protected void map(Text key, BytesWritable value, Context context) throws IOException, InterruptedException {
    context.write(key, value);
  }
}

@Slf4j
public class CustomArchiveReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> {

  @Override
  protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {
    context.write(key, values.iterator().next());
  }
}

public class CustomArchiveDriver {

  public static void main(String[] args) throws Exception {
    // 获取JOB
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "hdfs://192.168.1.161:9000");
    Job job = Job.getInstance(configuration);
    // 设置jar位置
    job.setJarByClass(CustomArchiveDriver.class);
    // 关联Mapper和Reducer
    job.setMapperClass(CustomArchiveMapper.class);
    job.setReducerClass(CustomArchiveReducer.class);
    job.setInputFormatClass(CustomArchiveInputFormat.class);
    // 设置Mapper阶段输出的KV类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(BytesWritable.class);
    // 设置最后输出的KV类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(BytesWritable.class);
    // 设置程序输出的输入路径和输出路径
    FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.1.161:9000/input"));
    FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.1.161:9000/output"));
    // 提交JOB
    job.submit();
    job.waitForCompletion(true);
  }
}

/**
 * 一下代码没用取模的方式，所以如果taskNum设置小于5，返回值会大于numPartitions
 * 分区号必须从0开始，逐一累加
 */
public class CustomPartition extends Partitioner<Text, FlowerBean> {
  @Override
  public int getPartition(Text text, FlowerBean flowerBean, int numPartitions) {
    String phone = text.toString();
    if(phone.startsWith("139")) {
      return 0;
    }
    if(phone.startsWith("183")) {
      return 1;
    }
    if(phone.startsWith("135")) {
      return 2;
    }
    if(phone.startsWith("137")) {
      return 3;
    }
    return 4;
  }
}

public class FlowerDriver {
  public static void main(String[] args) throws Exception {
    // 。。。
    job.setPartitionerClass(CustomPartition.class);
    job.setNumReduceTasks(5);
    // 。。。
  }
}

/**
 * 现象：reducer并未将价格加和
 * 原理：进入顺序
 *      <0000001,33.8> 
 *      <0000001,222.8>
 *      <0000002,122.4>
 *      ......
 *
 *    实际数据                      reducer结果
 * 0000001  222.8       ==>      0000001  33.8
 * 0000002  722.4                0000001  222.8
 * 0000001  33.8                 0000002  122.4
 * 0000003  232.8                0000002  522.8
 * 0000003  33.8                 0000002  722.4 
 * 0000002  522.8                0000003  33.8 
 * 0000002  122.4                0000003  232.8 
 */
@Data
public class OrderBean implements WritableComparable<OrderBean> {
  private int orderId; // 订单id号
  private double price; // 价格

  @Override
  public int compareTo(OrderBean o) {
    int orderIdCompare = Integer.compare(this.getOrderId(), o.getOrderId());
    return orderIdCompare == 0 ? Double.compare(this.getPrice(), o.getPrice()) : orderIdCompare;
  }

  @Override
  public void write(DataOutput out) throws IOException {
    out.writeInt(orderId);
    out.writeDouble(price);
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    orderId = in.readInt();
    price = in.readDouble();
  }
}

public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> {

  @Override
  protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
    OrderBean currentKey = new OrderBean();
    for (NullWritable value : values) {
      currentKey.setOrderId(key.getOrderId());
      currentKey.setPrice(currentKey.getPrice() + key.getPrice());
      currentKey.setDate(key.getDate());
    }
    context.write(currentKey, NullWritable.get());
  }
}

/**
 * 现象：reducer并将价格加和了
 * 原理：进入顺序
 *      <0000001,33.8,2020-09-14> <0000001,222,2020-09-14> 
 *      <0000002,122.4,2020-09-10>
 *      <0000002,522.8,2020-09-14>
 *      <0000002,722.4,2020-09-12>
 *      <0000003,33.8,2020-09-11> <0000003,232.8,2020-09-11>
 *      ......
 * 
 *    实际数据                                      reducer结果
 * 0000001 222    2020-09-14       ==>      0000001 255.8 2020-09-14
 * 0000002 722.4  2020-09-12                0000002 122.4 2020-09-10
 * 0000001 33.8   2020-09-14                0000002 522.8 2020-09-14
 * 0000003 232.8  2020-09-11                0000002 722.4 2020-09-12
 * 0000003 33.8   2020-09-11                0000003 266.6 2020-09-11
 * 0000002 522.8  2020-09-14
 * 0000002 122.4  2020-09-10
 */
@Data
public class OrderBean implements WritableComparable<OrderBean> {

  private int orderId;
  private double price;
  private String date;

  @Override
  public int compareTo(OrderBean o) {
    int orderIdCompare = Integer.compare(this.getOrderId(), o.getOrderId());
    return orderIdCompare == 0 ? Double.compare(this.getPrice(), o.getPrice()) : orderIdCompare;
  }

  @Override
  public void write(DataOutput out) throws IOException {
    out.writeInt(orderId);
    out.writeDouble(price);
    out.writeUTF(date);
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    orderId = in.readInt();
    price = in.readDouble();
    date = in.readUTF();
  }
}

public class OrderGroupingComparator extends WritableComparator {

  public OrderGroupingComparator() {
    super(OrderBean.class,true);
  }

  @Override
  public int compare(WritableComparable a, WritableComparable b) {
    OrderBean aBean = (OrderBean) a;
    OrderBean bBean = (OrderBean) b;
    return aBean.getDate().compareTo(bBean.getDate());
  }
}

// 获得数据按照orderId分组price排序
public class OrderReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> {

  @Override
  protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
    OrderBean currentKey = new OrderBean();
    for (NullWritable value : values) {
      currentKey.setOrderId(key.getOrderId());
      currentKey.setPrice(currentKey.getPrice() + key.getPrice());
    }
    context.write(currentKey, NullWritable.get());
  }
}

public class LogOutputFormat extends FileOutputFormat<Text, NullWritable> {
  @Override
  public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
    return new LogRecordWriter(job);
  }
}

/**
 * 将数据根据行首字母分类
 */
public class LogRecordWriter extends RecordWriter<Text, NullWritable> {

  private FSDataOutputStream outA_Q;
  private FSDataOutputStream outR_Z;

  public LogRecordWriter(TaskAttemptContext job) {
    try {
      Configuration configuration = job.getConfiguration();
      FileSystem fileSystem = FileSystem.get(configuration);
      String dir = configuration.get(FileOutputFormat.OUTDIR);
      outA_Q = fileSystem.create(new Path(dir + "/" + "a-q.text"));
      outR_Z = fileSystem.create(new Path(dir + "/" + "r-z.text"));
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  @Override
  public void write(Text key, NullWritable value) throws IOException, InterruptedException {
    String keyStr = key.toString();
    if(StringUtils.isNotBlank(keyStr)) {
      char firstChar = keyStr.toLowerCase().charAt(0);
      if ('a' <= firstChar && firstChar <= 'q') {
        outA_Q.writeUTF(keyStr + "\n");
      } else {
        outR_Z.writeBytes(keyStr + "\n");
      }
    }
  }

  @Override
  public void close(TaskAttemptContext context) throws IOException, InterruptedException {
    outA_Q.close();
    outR_Z.close();
  }
}

@Data
public class TableBean implements Writable {
  private String orderId;
  private String productId;
  private String name;
  private int count;
  private String flag; // 表的标记

  @Override
  public void write(DataOutput out) throws IOException {
    out.writeUTF(orderId);
    out.writeUTF(productId);
    out.writeUTF(name);
    out.writeInt(count);
    out.writeUTF(flag);
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    orderId = in.readUTF();
    productId = in.readUTF();
    name = in.readUTF();
    count = in.readInt();
    flag = in.readUTF();
  }

  @Override
  public String toString() {
    return orderId + '\t' +  name + '\t' + count + '\t';
  }
}

// 分类
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {
  private String fileName;
  private TableBean tableBean = new TableBean();
  private Text pId = new Text();

  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    FileSplit inputSplit = (FileSplit)context.getInputSplit();
    fileName = inputSplit.getPath().getName();  
  }

  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String[] data = value.toString().split("\t");
    if(fileName.startsWith("order")) {
      tableBean.setOrderId(data[0]);
      tableBean.setProductId(data[1]);
      pId.set(data[1]);
      tableBean.setCount(Integer.parseInt(data[2]));
      tableBean.setName("");
      tableBean.setFlag("order");
      context.write(pId, tableBean);
    }

    if(fileName.startsWith("product")) {
      tableBean.setOrderId("");
      tableBean.setCount(0);
      tableBean.setProductId(data[0]);
      tableBean.setName(data[1]);
      pId.set(data[0]);
      tableBean.setFlag("product");
      context.write(pId, tableBean);
    }
  }
}

public class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {

  @Override
  protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {

    List<TableBean> orderBeans = new ArrayList<>();
    TableBean productBean = new TableBean();

    // 必须重新copy对象，不然iterator时使用自己填充，会将记录后移
    for (TableBean value : values) {
      if(value.getFlag().equals("order")) {
        TableBean bean = new TableBean();
        BeanUtil.copyProperties(value, bean);
        orderBeans.add(bean);
      } else {
        BeanUtil.copyProperties(value, productBean);
      }
    }

    for (TableBean orderBean : orderBeans) {
      TableBean record = new TableBean();
      BeanUtil.copyProperties(orderBean, record);
      record.setName(productBean.getName());
      context.write(record, NullWritable.get());
    }
  }
}

public class TableMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

  private Map<String, String> productMap = new HashMap<>();
  private Text k = new Text();

  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    // 1 获取缓存的文件
    URI[] cacheFiles = context.getCacheFiles();
    String path = cacheFiles[0].getPath();
    Configuration configuration = context.getConfiguration();
    FileSystem fileSystem = FileSystem.get(configuration);
    FSDataInputStream fileStream = fileSystem.open(new Path(path));
    BufferedReader reader = new BufferedReader(new InputStreamReader(fileStream));
    String line;
    while(StringUtils.isNotEmpty(line = reader.readLine())){
      // 2 切割
      String[] fields = line.split("\t");
      // 3 缓存数据到集合
      productMap.put(fields[0], fields[1]);
    }
    // 4 关流
    reader.close();
  }

  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    // 1 获取一行
    String line = value.toString();
    // 2 截取
    String[] fields = line.split("\t");
    // 3 获取产品id
    String pId = fields[1];
    // 4 获取商品名称
    String pdName = productMap.get(pId);
    // 5 拼接
    k.set(line + "\t"+ pdName);
    // 6 写出
    context.write(k, NullWritable.get());
  }
}

public class DistributedCacheDriver {

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", "hdfs://192.168.1.161:9000");

        Job job = Job.getInstance(configuration);
        // 设置缓存文件
        job.addCacheFile(new URI("hdfs://192.168.1.161:9000/join/product.txt"));

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setMapperClass(TableMapper.class);

        job.setJarByClass(DistributedCacheDriver.class);

        FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.1.161:9000/join"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.1.161:9000/output"));
        job.submit();
        job.waitForCompletion(true);
    }
}

/* 可使用字符串区分 */
context.getCounter("status", "success").increment(1);
context.getCounter("status", "fail").increment(1);

/* 可使用枚举 */
context.getCounter(HttpStatus.SUCCESS).increment(1);
context.getCounter(HttpStatus.FAIL).increment(1);

// 开启map端输出压缩
configuration.setBoolean("mapreduce.map.output.compress", true);
// 设置map端输出压缩方式
configuration.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

// 设置reduce端输出压缩开启
FileOutputFormat.setCompressOutput(job, true);
// 设置压缩的方式
FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); 

public class FlowerDriver implements Tool {

  private Configuration configuration;

  public static void main(String[] args) throws Exception {
    int res = ToolRunner.run(new FlowerDriver(), args);
    System.exit(res);
  }

  @Override
  public int run(String[] args) throws Exception {
    // 获取JOB
    Job job = Job.getInstance(this.configuration);
    // 关联Mapper和Reducer
    job.setMapperClass(FlowerMapper.class);
    job.setReducerClass(FlowerReducer.class);
    // 设置Mapper阶段输出的KV类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(FlowerBean.class);
    // 设置最后输出的KV类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(FlowerBean.class);
    // 设置程序输出的输入路径和输出路径
    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    // 提交JOB
    job.submit();
    return job.waitForCompletion(true) ? 0 : -1;
  }

  @Override
  public void setConf(Configuration conf) {
    this.configuration = conf;
  }

  @Override
  public Configuration getConf() {
    return configuration;
  }
}