<dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version> <!--2.6.0-cdh5.16.2-->
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13</version>
        </dependency>
    </dependencies>

  @Test
public void mkdir() throws Exception {
        Configuration configuration = new Configuration();

        URI uri = new URI("hdfs://ruozedata001:8020");

        // 获取文件系统客户端对象  mock用户
        FileSystem fileSystem = FileSystem.get(uri, configuration, "hadoop");

        // 对文件系统做相应的操作
        Path path = new Path("/hdfsapi");
        final boolean isSucceeded = fileSystem.mkdirs(path);

        // 断言
        Assert.assertEquals(true, isSucceeded);

        // 释放资源
        fileSystem.close();
    }

@Test
 public void CopyFromLocal() throws Exception {

     System.setProperty("HADOOP_USER_NAME", "bigdata");
     Configuration configuration = new Configuration();
     URI uri = new URI("hdfs://hadoop001:9000");
     FileSystem fileSystem = FileSystem.get(uri, configuration);

     Path src = new Path("data/wc.txt");
     Path dst = new Path("/hadoop/data/");
     fileSystem.copyFromLocalFile(src, dst);

     fileSystem.close();
 }

org.apache.hadoop.ipc.RemoteException(java.io.IOException): File /hadoop/data/wc.txt could only be replicated to 0 nodes instead of minReplication (=1).  There are 1 datanode(s) running and 1 node(s) are excluded in this operation.
	at org.apache.hadoop.hdfs.server.blockmanagement.BlockManager.chooseTarget4NewBlock(BlockManager.java:1723)
	at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:3508)
	at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.addBlock(NameNodeRpcServer.java:694)
	....

/***
     * 下载文件
     * @param src
     * @param dest
     * @throws Exception
     */
    public static void get(String src, String dest) throws Exception {
        FileSystem fileSystem = getFileSystem();
        //1. HDFS
        Path inputPath = new Path(src);
        //2.本地
        Path outputPath = new Path(dest);
        fileSystem.copyToLocalFile(false,inputPath, outputPath,true);
        fileSystem.close();
    }

public class FileUtils {
    /**
     * 获取FileSystem
     *
     * @return
     */
    static FileSystem getFileSystem() {
        System.setProperty(HADOOP_USER_NAME, "bigdata");
        Properties properties = ParamUtil.getProperties();
        Configuration configuration = new Configuration();
        configuration.set("dfs.client.use.datanode.hostname", "true");
        configuration.set("dfs.replication", "1");

        try {
            URI uri = new URI(properties.getProperty(HDFS_URI));
            FileSystem fileSystem = FileSystem.get(uri, configuration);
            return fileSystem;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
    
    /**
     * 删除目标文件
     * @param configuration
     * @param output
     */

    public static void deleteTarget(Configuration configuration, String output) {

        try {

            FileSystem fileSystem = FileSystem.get( configuration);

            boolean exists = fileSystem.exists(new Path(output));
            if (exists) {
                fileSystem.delete(new Path(output), true);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

1020 YarnChild
1023 YarnChild
9021 MRAppMaster

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

    IntWritable ONE = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 获取到内容
        final String line = value.toString().toLowerCase();

        // 按照分隔符进行拆分
        final String[] splits = line.split(",");

        // 输出
        for (String word : splits) {
            context.write(new Text(word), ONE);
        }

    }
}

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

    /**
     *
     * 排序：默认key 字典序
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        // 对每个key做聚合操作(次数相加)
        int count = 0;
        for (IntWritable value : values) {
            count += value.get();
        }

        context.write(key, new IntWritable(count));
    }
}

public class WordCountDriver {

    public static void main(String[] args) throws Exception {
        String input = "data/wc.txt";
        String output = "out";

        final Configuration configuration = new Configuration();

        // 1、获取Job对象
        final Job job = Job.getInstance(configuration);

        // 删除输出文件夹
        FileUtils.deleteTarget(configuration,output);

        // 2、设置class
        job.setJarByClass(WordCountDriver.class);

        // 3、设置Mapper和Reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        // 4、设置Mapper阶段输出数据的key和value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 5、设置Reducer阶段输出数据的key和value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 6、设置输入和输出数据的路径
        FileInputFormat.setInputPaths(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        // 7、提交Job
        final boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

public void run(Context context) throws IOException, InterruptedException {
    setup(context);
    try {
      while (context.nextKeyValue()) {
        map(context.getCurrentKey(), context.getCurrentValue(), context);
      }
    } finally {
      cleanup(context);
    }
  }

public class Access  implements Writable {

    private String phone;
    private long up;
    private long down;
    private long sum;

    // 序列化
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(phone);
        out.writeLong(up);
        out.writeLong(down);
        out.writeLong(sum);
    }

    // 反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        phone = in.readUTF();
        up = in.readLong();
        down = in.readLong();
        sum = in.readLong();
    }

    @Override
    public String toString() {
        return "Access{" +
                "phone='" + phone + '\'' +
                ", up=" + up +
                ", down=" + down +
                ", sum=" + sum +
                '}';
    }

	//getter setter方法略
}

public class AccessDriver {
    public static void main(String[] args) throws Exception {
        String input = "data/access.log";
        String output = "out";

         Configuration configuration = new Configuration();

        // 1、获取Job对象
         Job job = Job.getInstance(configuration);

        // 删除输出文件夹
       FileUtils.deleteTarget(configuration,output);

        // 2、设置class
        job.setJarByClass(AccessDriver.class);

        // 3、设置Mapper和Reducer
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        // 4、设置Mapper阶段输出数据的key和value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Access.class);

        // 5、设置Reducer阶段输出数据的key和value的类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Access.class);

        // 6、设置输入和输出数据的路径
        FileInputFormat.setInputPaths(job, new Path(input));
        FileOutputFormat.setOutputPath(job, new Path(output));

        // 7、提交Job
        final boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

    public static class MyMapper extends Mapper<LongWritable, Text, Text, Access> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] splits = value.toString().split("\t");
            final String phone = splits[1];  // 第二个字段：手机号
            long up = Long.parseLong(splits[splits.length-3]);  //倒数第三个字段：上行流量
            long down = Long.parseLong(splits[splits.length-2]); // 倒数第二个字段：下行流量

            final Access access = new Access();
            access.setPhone(phone);
            access.setUp(up);
            access.setDown(down);
            access.setSum(up + down);

            context.write(new Text(phone), access);
        }
    }

    public static class MyReducer extends Reducer<Text, Access, NullWritable, Access> {
        @Override
        protected void reduce(Text phone, Iterable<Access> values, Context context) throws IOException, InterruptedException {
            long ups = 0;
            long downs = 0;

            for (Access access : values) {
                ups += access.getUp();
                downs += access.getDown();
            }

            final Access access = new Access();
            access.setPhone(phone.toString());
            access.setUp(ups);
            access.setDown(downs);
            access.setSum(ups + downs);

            context.write(NullWritable.get(), access);
        }
    }
}

job.waitForCompletion{
	submit() {  
		connect() {
			// 1)本地  2)YARN
			return new Cluster(getConfiguration()); 
		}
		
		JobSubmitter submitter // 作业提交器
		submitter.submitJobInternal(Job.this, cluster) {// *****
		//1. 数据输出目录是否存在
		//2. 输出有无配置
			checkSpecs(job);
		//submitJobDir存放作业相关的信息
			Path submitJobDir = new Path(jobStagingArea, jobId.toString());
			//拷贝
			copyAndConfigureFiles(job, submitJobDir);
			
			// job.xml维护job的相关信息
			Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
			//计算Map(InputSplit)数量
			int maps = writeSplits(job, submitJobDir); {
				minSize = max(1,0) = 1
				maxSize = Long.MAX_VALUE
				
				blockSize = 32M
				//计算splitSize
				splitSize = computeSplitSize(blockSize, minSize, maxSize);{
					Math.max(minSize, Math.min(maxSize, blockSize));
					max(1, min(Long.MAX_VALUE, 32M))
					max(1, 32M) = 32M
				}
				//SPLIT_SLOP=1.1		((double)bytesRemaining)/splitSize>SPLIT_SLOP
			
			}
			//配置文件写入
			writeConf(conf, submitJobFile);
		}
		
	}
}

public abstract 
    List<InputSplit> getSplits(JobContext context
                               ) throws IOException, InterruptedException;
      
public abstract 
    RecordReader<K,V> createRecordReader(InputSplit split,
                                         TaskAttemptContext context
                                        ) throws IOException, 
                                                 InterruptedException;

 final Configuration configuration = new Configuration();
 //在job获取之前设置才生效
configuration.set(KEY_VALUE_SEPERATOR,",");

final Job job = Job.getInstance(configuration);

public static class MyMapper extends Mapper<Text, Text, Text, IntWritable> {
        IntWritable ONE = new IntWritable(1);

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

public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        @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));
        }
    }

public class AccessPartitioner extends Partitioner<Text, Access> {

    @Override
    public int getPartition(Text text, Access access, int numPartitions) {
        final String telephone = text.toString();
        if(telephone.startsWith("13")) {
            return  0;
        } else if (telephone.startsWith("15")) {
            return 1;
        } else {
            return 2;
        }
    }
}

public interface WritableComparable<T> extends Writable, Comparable<T> {
}

public class Access implements WritableComparable<Access> {
	//省略序列化方法...

 	// 自定义排序方法
    @Override
    public int compareTo(Access o) {
        return this.getSum() > o.getSum() ? 1 : -1;
}

public static class MyMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            final String line = value.toString().toLowerCase();
            final String[] splits = line.split(",");

            for (String word : splits) {
                context.write(new Text(word), NullWritable.get());
            }
        }
    }

public static class MyReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            context.write(key, NullWritable.get());
        }
    }

public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        IntWritable ONE = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            final String line = value.toString().toLowerCase();
            final String[] splits = line.split("\t");

            // 代码容错
            if(splits.length == 8) {
                context.write(new Text(splits[7].trim()), ONE);
            }
        }
    }

public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            // 对每个key做聚合操作(次数相加)
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }

            context.write(key, new IntWritable(count));
        }
    }

public static class MyMapper extends Mapper<LongWritable, Text, IntWritable, Info> {
        String name;

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

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] splits = value.toString().split("\t");

            // 在同一个map中如何区分出来这一行数据到底是来自emp还是dept
            if (name.contains("emp")) {  // emp
                final int empno = Integer.parseInt(splits[0].trim());
                final String ename = splits[1].trim();
                final int deptno = Integer.parseInt(splits[7].trim());

                Info info = new Info();
                info.setEmpno(empno);
                info.setEname(ename);
                info.setDeptno(deptno);
                info.setDname("");
                info.setFlag(1);

                context.write(new IntWritable(deptno), info);
            } else { // dept
                Info info = new Info();
                final int deptno = Integer.parseInt(splits[0].trim());
                final String dname = splits[1].trim();

                info.setDeptno(deptno);
                info.setDname(dname);
                info.setFlag(2);
                info.setEmpno(0);
                info.setEname("");

                context.write(new IntWritable(deptno), info);
            }
        }
    }

public static class MyReducer extends Reducer<IntWritable, Info, Info, NullWritable> {
        @Override
        protected void reduce(IntWritable key, Iterable<Info> values, Context context) throws IOException, InterruptedException {

            List<Info> infos = new ArrayList<>();

            String dname = "";

            // 真正完成join操作
            for (Info info : values) {
                if(info.getFlag() == 1) { // emp
                    Info tmp = new Info();
                    tmp.setEmpno(info.getEmpno());
                    tmp.setEname(info.getEname());
                    tmp.setDeptno(info.getDeptno());
                    infos.add(tmp);
                } else {  // dept
                    dname = info.getDname();
                }
            }


            for(Info bean : infos) {
                bean.setDname(dname);
                context.write(bean, NullWritable.get());
            }
        }
    }

Reduce input groups=4
Reduce shuffle bytes=511
Reduce input records=18
Reduce output records=14
Spilled Records=36
Shuffled Maps =2
Failed Shuffles=0

...
// 5、设置Reducer阶段输出数据的key和value的类型

 job.setNumReduceTasks(0);  // 明确告诉MR框架没有reducer

// 小表放到缓存中
job.addCacheFile(new URI("data/join/dept.txt"));

...

public static class MyMapper extends Mapper<LongWritable, Text, Info, NullWritable> {

        Map<String,String> cacheMap = new HashMap<>();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
        //获取缓存中的信息
            final String path = context.getCacheFiles()[0].getPath();

            final BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path), "UTF-8"));

            String line;
            while (StringUtils.isNotEmpty(line = reader.readLine())) {
                final String[] splits = line.split("\t");
                //deptno,dname
                cacheMap.put(splits[0], splits[1]);
            }

            IOUtils.closeStream(reader);

        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            final String[] splits = value.toString().split("\t");
            final int empno = Integer.parseInt(splits[0].trim());
            final String ename = splits[1].trim();
            final int deptno = Integer.parseInt(splits[7].trim());

            Info info = new Info();
            info.setEmpno(empno);
            info.setEname(ename);
            info.setDeptno(deptno);

            info.setDname(cacheMap.get(deptno+""));

            context.write(info, NullWritable.get());
        }
    }

Map-Reduce Framework
		Map input records=14
		Map output records=14
		Input split bytes=114
		Spilled Records=0
		Failed Shuffles=0
		Merged Map outputs=0