<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
        <version>2.7.4</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.7.4</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>2.7.4</version>
    </dependency>
</dependencies>

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * Created by chingy on 2017/9/29.
 */
public class MyWordCount {
    //Mapper class
    static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final static IntWritable ONE = new IntWritable(1);
        private Text word = new Text();

@Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//            super.map(key, value, context);//参考源码最基本的处理
            StringTokenizer itr = new StringTokenizer(value.toString());//默认根据空白符/t/n/r/f进行分割
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());//取出每个单词作为map输出的key
                context.write(word, ONE);//通过上下文输出map的key-value，value以1为计数
            }
        }
    }

//Reducer class
    static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

@Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
//            super.reduce(key, values, context);//参考源码最基本的处理
            int sum = 0;//用于累加
            for (IntWritable value : values) {//循环同key的value组，进行处理
                sum += value.get();//累加
            }
            result.set(sum);//累加结果作为reduce输出的value
            context.write(key, result);
        }
    }

//Driver
    public static void run(String[] args) throws Exception {
        Configuration conf = new Configuration();//获取运行环境下的配置信息
        conf.set("mapred.jar","D:/IDEAWorks/hadoopAPI/target/hadoopAPI-1.0-SNAPSHOT.jar");
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: MyWordCount <in> <out>");
            System.exit(2);
        }
        //创建Jog，设置配置及Job名称
//        Job job = new Job(configuration, "wc");//新版废弃的方法
        Job job = Job.getInstance(conf, "WC");
        //1·设置Job运行的类
        job.setJarByClass(MyWordCount.class);
        //2·设置Mapper及Reducer
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //3·设置输入、输出文件路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //4·提交Job，等待运行结果，并在客户端显示运行信息
        boolean isSuccess = job.waitForCompletion(true);
        //结束程序，并返回执行结果
        System.exit(isSuccess ? 0 : 1);
    }

public static void main(String[] args) throws Exception {
        args = new String[] { "file:///D:/data/taotaomm.txt", "hdfs://n1/remoteUser/mmResult" };
        run(args);//这里手动连接的NN，不知道应该怎么解决--undo
    }
}

hdfs dfs -mkdir /remoteUser

hdfs dfs -chown chingy:supergroup /remoteUser

hdfs dfs -ls /

远程创建目录，测试授权是否成功：

@Before
public void init() throws Exception {
    //配置文件
    Configuration conf = new Configuration();
    //最小化配置
    conf.set("fs.default.name", "hdfs://n1");
    //获取文件系统
    hdfs = FileSystem.get(conf);
}

@Test
public void mkdir() throws Exception {
    Path path = new Path("/remoteUser/chingy");
    if (hdfs.mkdirs(path))
        System.out.println("create dir success!");
}

public static boolean access(String path, AccessRight desiredAccess)
    throws IOException {
return true;
//return access0(path, desiredAccess.accessRight());
}

log4j.rootLogger=WARN,A1
log4j.appender.A1=org.apache.log4j.ConsoleAppender
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c]-[%p] %m%n

mapred-site.xml

<configuration>
    <property>
        <!-- 提交任务的方式：yarn -->
        <name>mapreduce.framework.name</name>
        <value>yarn</value>
    </property>
    <property>
        <!-- 开启跨平台，环境变量会得到自动处理，官方文档:将linux中的$PATH、windows中的%PATH%，统一设置为{{PATH}} -->
        <name>mapreduce.app-submission.cross-platform</name>
        <value>true</value>
    </property>
</configuration>

yarn-site.xml

<configuration>
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
    </property>
    <property>
        <!-- 指出提交任务的RM服务器主机名，其它地址都是该地址的间接引用,网上都是配置的yarn.resourcemanager.address -->
        <name>yarn.resourcemanager.hostname</name>
        <value>n1</value>
    </property>
</configuration>

hdfs-site.xml

<configuration>
    <property>
        <!-- 避免操作的文件备份数被重置为默认的3 -->
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>