大数据 - MapReduce 编程实践 - 《大数据实验》

一、添加配置
二、FileMerge
三、SortedData

一、添加配置

vi $HADOOP_HOME/etc/hadoop/mapred-site.xml

添加

    <property>
        <name>yarn.app.mapreduce.am.env</name>
        <value>HADOOP_MAPRED_HOME=${HADOOP_HOME}</value>
    </property>
    <property>
        <name>mapreduce.map.env</name>
        <value>HADOOP_MAPRED_HOME=${HADOOP_HOME}</value>
    </property>
    <property>
        <name>mapreduce.reduce.env</name>
        <value>HADOOP_MAPRED_HOME=${HADOOP_HOME}</value>
    </property>

二、FileMerge

2.1、准备测试数据

cd ~
mkdir filemerge
cd filemerge

echo '20150101 x
20150102 y
20150103 x
20150104 y
20150105 z
20150106 x' > A.txt

echo '20150101 y
20150102 y
20150103 x
20150104 z
20150105 y' > B.txt

2.2、上传到hdfs

hdfs dfs -mkdir -p input1
hdfs dfs -put A.txt B.txt input1

2.3、编写代码

vi FileMerge.java

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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;
/* https://blog.csdn.net/weixin_36708477/article/details/86237012 */
public class FileMerge {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();//获得程序参数 放到otherArgs中
        //   String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if(otherArgs.length < 2) {
            System.err.println("Usage: FileMerge <in> [<in>...] <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "FileMerge");//设置环境参数
        job.setJarByClass(FileMerge.class);//设置整个程序的类名
        job.setMapperClass(FileMerge.Map.class);//添加Mapper类
        job.setCombinerClass(FileMerge.Reduce.class);//添加combiner类
        job.setReducerClass(FileMerge.Reduce.class);//添加reducer类
        job.setOutputKeyClass(Text.class);//设置输出类型
        job.setOutputValueClass(Text.class);//设置输出类型
        //System.exit(job.waitForCompletion(true) ? 0 : 1);
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));//设置输入文件
        }
        Path outPutPath = new Path(otherArgs[otherArgs.length - 1]);
        FileOutputFormat.setOutputPath(job, outPutPath);//设置输出文件
        FileSystem fileSystem = outPutPath.getFileSystem(conf);
        if (fileSystem.exists(outPutPath)) {//如果存在output文件夹则删除
            fileSystem.delete(outPutPath, true);// true的意思是，就算output有东西，也一带删除
        }
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
    //实现map函数
    public static class Map extends Mapper<Object, Text, Text, Text> {//继承mapper父类  //接收一个文件名作为key，该文件的每行内容作为value
        private final Text emptyText = new Text();
        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            context.write(value, emptyText);//把每一行都放到text中 唯一的key实现了去重
        }
    }
    // reduce将输入中的key复制到输出数据的key上，并直接输出，这是数据去重思想
    public static class Reduce extends Reducer<Text, Text, Text, Text> {
        private final Text emptyText = new Text();
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            context.write(key, emptyText); //map传给reduce的数据已经做完数据去重，输出即可
        }
    }
}

2.4、编译

export CLASSPATH="/usr/local/hadoop/share/hadoop/common/hadoop-common-3.2.1.jar:/usr/local/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.2.1.jar:/usr/local/hadoop/share/hadoop/common/lib/commons-cli-1.2.jar:$CLASSPATH"

javac -encoding utf-8 FileMerge.java

jar -cvf FileMerge.jar *.class

2.5、运行

hadoop jar FileMerge.jar FileMerge input1 output1

Bug记录:org.apache.hadoop.hdfs.server.namenode.SafeModeException 解决方法：原因是namenode处在安全模式下不能删除，执行： hadoop dfsadmin -safemode leave 退出安全模式即可（参见：https://blog.csdn.net/qq_16018407/article/details/78914559）

2.6 查看结果

hdfs dfs -ls output1
hdfs dfs -cat output1/*

三、SortedData

3.1、准备测试数据

cd ~
mkdir sorteddata
cd sorteddata

echo '111
222
555
999
111
333
222
444
888
666
777' > A.txt

echo '99
88
55
11
44
77
22
33
66
66' > B.txt

echo '5
2
8
1
6
3
4
9
7' > C.txt

3.2、上传到hdfs

hdfs dfs -mkdir -p input2
hdfs dfs -put A.txt B.txt C.txt input2

3.3、编写代码

vi SortedData.java

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.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;


/**
 * 数字排序
 */
/* https://blog.csdn.net/someby/article/details/82951610 */
public class SortedData {
    /**
     * 使用Mapper将数据文件中的数据本身作为Mapper输出的key直接输出
     */

    public static class forSortedMapper extends Mapper<Object, Text, IntWritable, IntWritable> {
        private IntWritable keyValue = new IntWritable(); //存放key的值
        private final IntWritable one = new IntWritable(1);

        /**
         * {key: num} -> {num: 1}
         * @param key object
         * @param value 每行数字的字符串
         * @param context context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString(); //获取读取的值，转化为String
            keyValue.set(Integer.parseInt(line)); //将String转化为Int类型
            context.write(keyValue, one); //将每一条记录标记为（key，value） key--数字 value--出现的次数
            //每出现一次就标记为（number，1）
        }
    }


    /**
     * 使用Reducer将输入的key本身作为key直接输出
     * < num, [1,1] >
     */
    public static class forSortedReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
        private IntWritable position = new IntWritable(1); //存放名次

        @Override
        protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            for (IntWritable item : values) { //同一个数字可能出多次，就要多次并列排序
                context.write(position, key); //写入名次和具体数字
                System.out.println(position + "\t" + key);
                position = new IntWritable(position.get() + 1); //名次加1
            }
        }
    }


    public static void main(String[] args) throws Exception {


        Configuration conf = new Configuration(); //设置MapReduce的配置
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.out.println("Usage: SortedData <in> [<in>...] <out>");
            System.exit(2);
        }

        //设置作业
        //Job job = new Job(conf);
        Job job = Job.getInstance(conf, "SortedData");
        job.setJarByClass(SortedData.class);

        //设置处理map,reduce的类
        job.setMapperClass(forSortedMapper.class);
        job.setReducerClass(forSortedReducer.class);

        //设置输入输出格式的处理
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);

        //设定输入路径
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        //设定输出路径
        Path outPutPath = new Path(otherArgs[otherArgs.length - 1]);
        FileOutputFormat.setOutputPath(job, outPutPath);

        FileSystem fileSystem = outPutPath.getFileSystem(conf);
        if (fileSystem.exists(outPutPath)) {//如果存在output文件夹则删除
            fileSystem.delete(outPutPath, true);// true的意思是，就算output有东西，也一带删除
        }

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

}

3.4、编译

export CLASSPATH="/usr/local/hadoop/share/hadoop/common/hadoop-common-3.2.1.jar:/usr/local/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.2.1.jar:/usr/local/hadoop/share/hadoop/common/lib/commons-cli-1.2.jar:$CLASSPATH"

javac -encoding utf-8 SortedData.java

jar -cvf SortedData.jar *.class

3.5、运行

hadoop jar SortedData.jar SortedData input2 output2

3.6、查看结果

hdfs dfs -ls output2
hdfs dfs -cat output2/*