package cn.adoredu.hdfs;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.FileInputStream;
import java.net.URI;

public class HDFSTestClient {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://node-1:9000");
        // 设置客户的身份
//        System.setProperty("HADOOP_USER_NAME", "gp");
        FileSystem fs = FileSystem.get(conf);
        // 也可以通过如下方式指定文件系统类型并且设置用户身份
        // FileSystem fs = FileSystem.get(new URI("hdfs://node-1:9000"), conf, "gp");
        // 创建一个目录
//        fs.create(new Path("/helloByJava"));
        // 下载文件
//        fs.copyToLocalFile(new Path("/tmp/input/wordcount.txt"), new Path("/Users/gp/Desktop/tmp"));
        // 下载后会在对应路径下有wordcount.txt和一个标识成功的.wordcount.txt.crc文件

        // 流式操作，更底层的操作
        // 将本地文件上传到hdfs
        FSDataOutputStream outputStream = fs.create(new Path("/test.txt"), true);
        FileInputStream inputStream = new FileInputStream("/Users/gp/Desktop/tmp/test.txt");
        IOUtils.copy(inputStream, outputStream);

        // 关闭连接
        fs.close();
    }
}

package cn.adoredu.hdfs;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.junit.Before;
import org.junit.Test;

public class HdfsClient {

	FileSystem fs = null;

	@Before
	public void init() throws Exception {


		Configuration conf = new Configuration();
		// conf.set("fs.defaultFS", "hdfs://node-1:9000");
		/**
		 * 参数优先级： 1、客户端代码中设置的值 2、classpath下的用户自定义配置文件 3、然后是jar中默认配置
		 */
		// 获取一个hdfs的访问客户端
		fs = FileSystem.get(new URI("hdfs://node-1:9000"), conf, "gp");

	}

	/**
	 * 往hdfs上传文件
	 * 
	 * @throws Exception
	 */
	@Test
	public void testAddFileToHdfs() throws Exception {

		// 要上传的文件所在的本地路径
		// 要上传到hdfs的目标路径
		Path src = new Path("/tmp/test.txt");
		Path dst = new Path("/");
		fs.copyFromLocalFile(src, dst);

		fs.close();
	}

	/**
	 * 从hdfs中复制文件到本地文件系统
	 * 
	 * @throws IOException
	 * @throws IllegalArgumentException
	 */
	@Test
	public void testDownloadFileToLocal() throws IllegalArgumentException, IOException {

		// fs.copyToLocalFile(new Path("/test.txt"), new Path("/tmp"));
		fs.copyToLocalFile(false, new Path("/test.txt"), new Path("/tmp"), true);
		fs.close();

	}

	/**
	 * 目录操作
	 * 
	 * @throws IllegalArgumentException
	 * @throws IOException
	 */
	@Test
	public void testMkdirAndDeleteAndRename() throws IllegalArgumentException, IOException {

		// 创建目录
		fs.mkdirs(new Path("/a1/b1/c1"));

		// 删除文件夹 ，如果是非空文件夹，参数2必须给值true
		fs.delete(new Path("/aaa"), true);

		// 重命名文件或文件夹
		fs.rename(new Path("/a1"), new Path("/a2"));

	}

	/**
	 * 查看目录信息，只显示文件
	 * 
	 * @throws IOException
	 * @throws IllegalArgumentException
	 * @throws FileNotFoundException
	 */
	@Test
	public void testListFiles() throws FileNotFoundException, IllegalArgumentException, IOException {
		
		RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

		while (listFiles.hasNext()) {

			LocatedFileStatus fileStatus = listFiles.next();

			System.out.println(fileStatus.getPath().getName());
			System.out.println(fileStatus.getBlockSize());
			System.out.println(fileStatus.getPermission());
			System.out.println(fileStatus.getLen());
			BlockLocation[] blockLocations = fileStatus.getBlockLocations();
			for (BlockLocation bl : blockLocations) {
				System.out.println("block-length:" + bl.getLength() + "--" + "block-offset:" + bl.getOffset());
				String[] hosts = bl.getHosts();
				for (String host : hosts) {
					System.out.println(host);
				}

			}

			System.out.println("--------------打印的分割线--------------");

		}

	}

	/**
	 * 查看文件及文件夹信息
	 * 
	 * @throws IOException
	 * @throws IllegalArgumentException
	 * @throws FileNotFoundException
	 */
	@Test
	public void testListAll() throws FileNotFoundException, IllegalArgumentException, IOException {

		FileStatus[] listStatus = fs.listStatus(new Path("/"));

		String flag = "";
		for (FileStatus fstatus : listStatus) {

			if (fstatus.isFile()) {
				flag = "f-- ";
			} else {
				flag = "d-- ";
			}
			System.out.println(flag + fstatus.getPath().getName());
			System.out.println(fstatus.getPermission());
		}
	}
}

package cn.adoredu.mapreduce;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * MR程序mapper阶段的处理类：
 *
 * Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 * KEYIN：mapper数据输入的key类型，默认的数据读取组件会一行一行的读取数据，
 *       读取一行返回一行。这里表示每一行的起始偏移量，数据类型为Long；
 * VALUEIN：这里表示读取没一行的内容，数据类型为String；
 * KEYOUT：本地输出的是单词，数据类型为String；
 * VALUEOUT：本地为单词次数，数据类型为Integer；
 *
 * 注意，这里的数据类型都为jdk的自带类型，跨网络传输序列化时效率地下，因此hadoop封装了对应的数据类型：
 * Long -> LongWritable;
 * String -> Text;
 * Integer -> Intwritable;
 * null -> NullWritable
 *
 * Mapper生命周期方法：
 * setup(): Called once at the beginning of the task
 * cleanup(): Called once at the end of the task
 * setup(): Called once for each key/value pair in the input split.
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    /**
     * mapper阶段的业务逻辑实现方法。该方法的调用取决于读取数据组件有无给mr传入数据。
     * 如果有，每传入一个kv对都调用一次。
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 拿到传入的一行内容，转化为String
        String line = value.toString();
        // 按照分隔符切割成单词数组
        String[] words = line.split(" ");
        // 遍历数组，每个单词标记1（整理交由reduce阶段）
        for (String word : words) {
            // 使用mr程序上下问context
            // 把mapper阶段处理的数据发送作为reduce阶段的输入
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

package cn.adoredu.mapreduce;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * MR程序reduce阶段的处理类：
 *
 * KEYIN：reduce阶段输入key类型，对应mapper阶段输出key类型，这里就是单词 Text；
 * VALUEIN：reduce阶段输入value类型，对应mapper阶段输出value类型，这里是单词次数， IntWritable；
 * KEYOUT：单词，Text；
 * VALUEOUT：单词总次数，IntWritable；
 *
 * Reducer生命周期方法：与Mapper中类似。
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    /**
     * reduce接收所有来自mapper阶段的数据，按照key字典序排序。然后按照key是否相同调用reduce方法。
     * 本方法把key作为参数key，所有v作为迭代器作为values。
     * @param key
     * @param values
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int count = 0;
        // 遍历一组迭代器
        for (IntWritable value : values) {
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

package cn.adoredu.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 程序运行的主类，封装程序执行的信息
 */
public class WordCountDriver {

    public static void main(String[] args) throws Exception {
        // 通过Job封装本次mr执行的信息
        Configuration conf = new Configuration();
        Job job = Job.getInstance();

        // 指定运行主类
        job.setJarByClass(WordCountDriver.class);

        // 指定mapper和reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        // 指定mapper阶段的输出k/v类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 指定最终输出的k/v类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 指定输入和输出数据位置
        FileInputFormat.setInputPaths(job, "/wordcount/input");
        FileOutputFormat.setOutputPath(job, new Path("/wordcount/output"));

        // 提交
//        job.submit();
        boolean res = job.waitForCompletion(true);  // 使用该方法可以打印执行日志
        System.exit(res?0:1);
    }
}

package cn.adoredu.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 程序运行的主类，封装程序执行的信息
 */
public class WordCountDriver {

    public static void main(String[] args) throws Exception {
        // 通过Job封装本次mr执行的信息
        Configuration conf = new Configuration();
        // 是否本地执行取决于该配置信息，值为local时表示本地执行
        conf.set("mapreduce.framework.name", "local");  // 实际上该配置可以省略，因为mapred-defalut.xml中默认就是local
        Job job = Job.getInstance();

        // 指定运行主类
        job.setJarByClass(WordCountDriver.class);

        // 指定mapper和reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        // 指定mapper阶段的输出k/v类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 指定最终输出的k/v类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 指定输入和输出数据位置
        // 本地执行，使用本地fs
        FileInputFormat.setInputPaths(job, "/Users/gp/Desktop/tmp/input");
        FileOutputFormat.setOutputPath(job, new Path("/Users/gp/Desktop/tmp/output"));

        // 提交
//        job.submit();
        boolean res = job.waitForCompletion(true);  // 使用该方法可以打印执行日志
        System.exit(res?0:1);
    }

}

package org.apache.hadoop.mapred.lib;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.mapred.Partitioner;
import org.apache.hadoop.mapred.JobConf;

/** 
 * Partition keys by their {@link Object#hashCode()}. 
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> {

  public void configure(JobConf job) {}

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K2 key, V2 value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

}
// & Integer.MAX_VALUE为了保证为正数。

public interface Writable {
 void write(DataOutput out) throws IOException;
 void readFields(DataInput in) throws IOException;
}

package cn.adoredu.flowsum;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class FlowBean implements Writable {

    private long upFlow;
    private long downFlow;
    private long sumFlow;

    public long getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }

    public long getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }

    public long getSumFlow() {
        return sumFlow;
    }

    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }

    // 序列化方法
    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    // 反序列化方法。注意反序列化时一定要和序列化时顺序一致
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }
   
}

package cn.adoredu.flowsum;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {

    // 提前定义，以免每次map都要新创建对象
    Text k = new Text();
    FlowBean flowBean = new FlowBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split("\t");

        String phoneNum = fields[1];
        long upFlow = Long.parseLong(fields[fields.length - 3]);
        long downFlow = Long.parseLong(fields[fields.length - 2]);

        k.set(phoneNum);
        flowBean.set(upFlow, downFlow);

        context.write(k, flowBean);
    }
}

package cn.adoredu.flowsum;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {

    FlowBean flowBean = new FlowBean();
  
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        long upFlowCount = 0;
        long downFlowCount = 0;

        for (FlowBean flowBean : values) {
            upFlowCount += flowBean.getUpFlow();
            downFlowCount += flowBean.getDownFlow();
        }
        flowBean.set(upFlowCount, downFlowCount);

        context.write(key, flowBean);
    }
}

package cn.adoredu.flowsum;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FlowSumDriver {

    public static void main(String[] args) throws Exception {
        // 通过Job封装本次mr执行的信息
        Configuration conf = new Configuration();
        // 是否本地执行取决于该配置信息，值为local时表示本地执行
        conf.set("mapreduce.framework.name", "local");  // 实际上该配置可以省略，因为mapred-defalut.xml中默认就是local
        Job job = Job.getInstance();

        // 指定运行主类
        job.setJarByClass(FlowSumDriver.class);

        // 指定mapper和reducer
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);

        // 指定mapper阶段的输出k/v类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        // 指定最终输出的k/v类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        // 指定输入和输出数据位置
        // 本地执行，使用本地fs
        FileInputFormat.setInputPaths(job, "/Users/gp/Desktop/tmp/flowsum/input");
        FileOutputFormat.setOutputPath(job, new Path("/Users/gp/Desktop/tmp/flowsum/output"));

        // 提交
//        job.submit();
        boolean res = job.waitForCompletion(true);  // 使用该方法可以打印执行日志
        System.exit(res?0:1);
    }
}

package org.apache.hadoop.io;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;

@InterfaceAudience.Public
@InterfaceStability.Stable
public interface WritableComparable<T> extends Writable, Comparable<T> {
}

package cn.adoredu.flowsum;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class FlowBean implements WritableComparable<FlowBean> {

    private long upFlow;
    private long downFlow;
    private long sumFlow;

    public FlowBean() {
    }

    public FlowBean(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = upFlow + downFlow;
    }

    public void set(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = upFlow + downFlow;
    }

    public long getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }

    public long getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }

    public long getSumFlow() {
        return sumFlow;
    }

    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }

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

    // 序列化方法
    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    // 反序列化方法。注意反序列化时一定要和序列化时顺序一致
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }

    // 为了实现倒序
    public int compareTo(FlowBean o) {
        return sumFlow>o.getSumFlow()?-1:1;
    }

}

package cn.adoredu.flowsum;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class FlowSumSort {

    public static class FlowSumSortMapper extends Mapper<LongWritable, Text, FlowBean, Text> {
        // 提前定义，以免每次map都要新创建对象
        Text v = new Text();
        FlowBean k = new FlowBean();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            String phoneNum = fields[1];
            long upFlow = Long.parseLong(fields[fields.length - 3]);
            long downFlow = Long.parseLong(fields[fields.length - 2]);

            v.set(phoneNum);
            k.set(upFlow, downFlow);

            context.write(k, v);  // 输出时kv和上次相反
        }
    }

    public static class FlowSumSortReducer extends Reducer<FlowBean, Text, Text, FlowBean> {
        @Override
        protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            context.write(values.iterator().next(), key); // reducer只作对调后输出
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 通过Job封装本次mr执行的信息
        Configuration conf = new Configuration();
        // 是否本地执行取决于该配置信息，值为local时表示本地执行
        conf.set("mapreduce.framework.name", "local");  // 实际上该配置可以省略，因为mapred-defalut.xml中默认就是local
        Job job = Job.getInstance();

        // 指定运行主类
        job.setJarByClass(FlowSumSort.class);

        // 指定mapper和reducer
        job.setMapperClass(FlowSumSortMapper.class);
        job.setReducerClass(FlowSumSortReducer.class);

        // 指定mapper阶段的输出k/v类型
        job.setMapOutputKeyClass(FlowBean.class);
        job.setMapOutputValueClass(Text.class);

        // 指定最终输出的k/v类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        // 指定输入和输出数据位置
        // 本地执行，使用本地fs
        FileInputFormat.setInputPaths(job, "/Users/gp/Desktop/tmp/flowsum/output");
        FileOutputFormat.setOutputPath(job, new Path("/Users/gp/Desktop/tmp/flowsum/outputsort"));

        // 提交
//        job.submit();
        boolean res = job.waitForCompletion(true);  // 使用该方法可以打印执行日志
        System.exit(res?0:1);
    }
}

package cn.adoredu.flowsum.partitioner;

import cn.adoredu.flowsum.FlowBean;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

import java.util.HashMap;

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {

    public static HashMap<String, Integer> provinceMap = new HashMap<String, Integer>();

    // 模拟省份划分规则
    static {
        provinceMap.put("131", 0);
        provinceMap.put("132", 1);
    }

    public int getPartition(Text text, FlowBean flowBean, int numPartitions) {

        Integer code = provinceMap.get(text.toString().substring(0, 3));

        if (code != null) {
            return code;
        }
        return 2;
    }
}

// ...
job.setNumReduceTasks(3);
// 设置分区规则
job.setPartitionerClass(ProvincePartitioner.class);
// ...

job.setCombinerClass(WordCountReducer.class);