MR-WordCount

pom.xml

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

    </dependencies>
    <build>
        <plugins>

            <!--主函数入口-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>2.4</version>
                <configuration>
                    <archive>
                        <manifest>
                            <addClasspath>true</addClasspath>
                            <classpathPrefix>lib/</classpathPrefix>
                            <mainClass>com.mr.demo.wordcount.WordCount</mainClass>
                        </manifest>
                    </archive>
                </configuration>
            </plugin>

            <!--jdk定义-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
        </plugins>
    </build>

WordCount.java

MapReduce编程案例
package com.flink.mr.demo.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

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

public class NeoWordCount {
    

    public static class NeoWordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    
        private final LongWritable ONE = new LongWritable(1);
        private final Text outputK = new Text();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
            for (String s : value.toString().split(" ")) {
    
                outputK.set(s);
                context.write(outputK, ONE);
            }
        }
    }
    public static class NeoWordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    
        private final LongWritable outputV = new LongWritable();
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
    
            long sum = 0;
            for (LongWritable value : values) {
    
                sum += value.get();
            }
            outputV.set(sum);
            context.write(key, outputV);
        }
    }


    public static void main(String[] args) throws Exception {
    
        /*GenericOptionsParser parser = new GenericOptionsParser(args);
        Job job = Job.getInstance(parser.getConfiguration());
        args = parser.getRemainingArgs();*/

        System.setProperty("HADOOP_USER_NAME","bigdata");
        Configuration config = new Configuration();

        config.set("fs.defaultFS","hdfs://10.1.1.1:9000");
        config.set("mapreduce.framework.name","yarn");
        config.set("yarn.resourcemanager.hostname","10.1.1.1");
        // 跨平台参数
        config.set("mapreduce.app-submission.cross-platform","true");

        Job job = Job.getInstance(config);
        job.setJar("D:\\bigdata\\mapreduces\\flink-mr.jar");


        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputKeyClass(LongWritable.class);

        job.setMapperClass(NeoWordCountMapper.class);
        job.setReducerClass(NeoWordCountReducer.class);
        job.setCombinerClass(NeoWordCountReducer.class);

        Path inputPath = new Path("/user/bigdata/demo/001/input");
        FileInputFormat.setInputPaths(job, inputPath);


        Path outputPath = new Path("/user/bigdata/demo/001/output");
        FileSystem fs = FileSystem.get(new URI("hdfs://10.1.1.1:9000"),config,"bigdata");
        if(fs.exists(outputPath)){
    
            fs.delete(outputPath,true);
        }
        FileOutputFormat.setOutputPath(job, outputPath);

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

}

环境要求
1.本地打包形成
D:\bigdata\mapreduces\flink-mr.jar

2.Hadoop环境
10.1.1.1

3.文件准备
hdfs://10.1.1.1:9000/user/bigdata/demo/001/input
上传几个文件用于分析

4.运行本示例，提交MR任务到集群