第四次

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-mapreduce-client-core</artifactId>
  <version>${hadoop.version}</version>
</dependency>

package cn.edu.nnu.mr;

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

import java.io.IOException;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    @Override
    protected void map(
            LongWritable key,  // 读到内容的位置
            Text value,        // 读到的内容
            Mapper<LongWritable, Text, Text, LongWritable>.Context context  // 输出
    ) throws IOException, InterruptedException {
        String[] ss = value.toString().split("[^A-Za-z]");  // 按照非英文字符进行分割
        LongWritable one = new LongWritable(1);
        for (String s : ss) {
            if (!s.trim().equals("")) {  // 输出非空的字符串
                context.write(
          new Text(s.trim()),  // KEYOUT
          one           // VALUEOUT
        );
            }
        }
    }
}

package cn.edu.nnu.mr;

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

import java.io.IOException;

public class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    @Override
    protected void reduce(
            Text key,  // 参数
            Iterable<LongWritable> values,  // 迭代器
            Reducer<Text, LongWritable, Text, LongWritable>.Context context  // 输出
    ) throws IOException, InterruptedException {
        long sum = 0L;
        for (LongWritable lw : values) {
            sum += lw.get();
        }
        context.write(
                key,  // KEYOUT
                new LongWritable(sum)  // VALUEOUT
        );
    }
}

package cn.edu.nnu.mr;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCountApp {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        // 创建任务
        Job job = Job.getInstance(conf);
        // 设置任务的主程序、Mapper 类、Reducer 类
        job.setJarByClass(WordCountApp.class);
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        // 设置输出数据的键、值类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        // 设置任务的输入、输出目录
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 执行任务
        try {
            job.waitForCompletion(true);
        } catch (Exception ex) {
            System.out.println(ex);
        }
    }
}

// 该部分为封装过的方法
mkdir("/test1127");  // 创建目录
put("D:\\iso\\data\\weather.txt", "/test1127/weather.txt");  // 上传文件

/* File: WeatherMapper */
package cn.edu.nnu.mr;

import ...

public class WeatherMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    @Override
    protected void map(
            LongWritable key,  // 读到内容的位置
            Text value,        // 读到的内容
            Mapper<LongWritable, Text, Text, LongWritable>.Context context  // 输出
    ) throws IOException, InterruptedException {
        String[] ss = value.toString().split(",");  // 按照 "," 拆分
        if (ss[0].equals("DATE")) {
            return;
        }
        // 日期作为 key，温度为 value
        try {
            context.write(
                    new Text(ss[0]),
                    new LongWritable(Long.parseLong(ss[5]))
            );
        } catch (Exception ex) {
        }
    }
}

/* File: WeatherReducer */
package cn.edu.nnu.mr;

import ...

public class WeatherReducer extends Reducer<Text, LongWritable, Text, DoubleWritable> {
    @Override
    protected void reduce(
            Text key,  // 参数
            Iterable<LongWritable> values,  // 迭代器
            Reducer<Text, LongWritable, Text, DoubleWritable>.Context context  // 输出
    ) throws IOException, InterruptedException {
        long sum = 0L;
        long cnt = 0L;
        for (LongWritable lw : values) {
            cnt++;
            sum += lw.get();
        }
        context.write(
                key,  // KEYOUT
                new DoubleWritable(sum * 1.0 / cnt)  // VALUEOUT
        );
    }
}

package cn.edu.nnu.mr;

import ...

public class WeatherApp {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        // 创建任务
        Job job = Job.getInstance(conf);
        // 设置任务的主程序、Mapper 类、Reducer 类
        job.setJarByClass(WeatherApp.class);
        job.setMapperClass(WeatherMapper.class);
        job.setReducerClass(WeatherReducer.class);
        // 设置输出数据的键、值类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        // 设置任务的输入、输出目录
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 执行任务
        try {
            job.waitForCompletion(true);
        } catch (Exception ex) {
            System.out.println(ex);
        }
    }
}

/* File: MonthWeatherMapper */
package cn.edu.nnu.mr;

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

import java.io.IOException;

public class MonthWeatherMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    @Override
    protected void map(
            LongWritable key,  // 读到内容的位置
            Text value,        // 读到的内容
            Mapper<LongWritable, Text, Text, LongWritable>.Context context  // 输出
    ) throws IOException, InterruptedException {
        String[] ss = value.toString().split(",");  // 按照 "," 拆分
        if (ss[0].equals("DATE")) {
            return;
        }
        // 月份作为 key，温度为 value
        String yymm = ss[0].substring(0, 6);  // 取前 6 位字串
        try {
            context.write(
                    new Text(yymm),  // 年月 201601
                    new LongWritable(Long.parseLong(ss[5]))  // 温度
            );
        } catch (Exception ex) {
        }
    }
}

/* File: MonthWeatherReducer */
package cn.edu.nnu.mr;

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

import java.io.IOException;

public class MonthWeatherReducer extends Reducer<Text, LongWritable, Text, DoubleWritable> {
    @Override
    protected void reduce(
            Text key,  // 参数
            Iterable<LongWritable> values,  // 迭代器
            Reducer<Text, LongWritable, Text, DoubleWritable>.Context context  // 输出
    ) throws IOException, InterruptedException {
        long sum = 0L;
        long cnt = 0L;
        for (LongWritable lw : values) {
            cnt++;
            sum += lw.get();
        }
        context.write(
                key,  // KEYOUT
                new DoubleWritable(sum * 1.0 / cnt)  // VALUEOUT
        );
    }
}

/* File: MonthWeatherApp */
package cn.edu.nnu.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
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;

import java.io.IOException;

public class MonthWeatherApp {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        // 创建任务
        Job job = Job.getInstance(conf);
        // 设置任务的主程序、Mapper 类、Reducer 类
        job.setJarByClass(MonthWeatherApp.class);  // 修改类名
        job.setMapperClass(MonthWeatherMapper.class);
        job.setReducerClass(MonthWeatherReducer.class);
        // 设置输出数据的键、值类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        // 设置任务的输入、输出目录
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 执行任务
        try {
            job.waitForCompletion(true);
        } catch (Exception ex) {
            System.out.println(ex);
        }
    }
}

<dependency>
  <groupId>com.huaban</groupId>
  <artifactId>jieba-analysis</artifactId>
  <version>1.0.2</version>
</dependency>

package demo;

import com.huaban.analysis.jieba.JiebaSegmenter;
import com.huaban.analysis.jieba.SegToken;

import java.util.List;

public class Test3 {
    public static void main(String[] args) {
        String str = "老师好我叫何同学";
        JiebaSegmenter jieba = new JiebaSegmenter();
        List<SegToken> tokens = jieba.process(str, JiebaSegmenter.SegMode.INDEX);
        for (SegToken st: tokens) {
            System.out.println(st.word);
        }
    }
}

/home/lc/hadoop/share/hadoop/common/lib/

put("D:\\iso\\data\\chinese.txt", "/test1127/chinese.txt");

/* File: CnWordCountMapper */
package cn.edu.nnu.mr;

import ...

public class CnWordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    @Override
    protected void map(
            LongWritable key,  // 读到内容的位置
            Text value,        // 读到的内容
            Mapper<LongWritable, Text, Text, LongWritable>.Context context  // 输出
    ) throws IOException, InterruptedException {
        JiebaSegmenter jieba = new JiebaSegmenter();
        List<SegToken> tokens = jieba.process(value.toString(), JiebaSegmenter.SegMode.INDEX);
        LongWritable one = new LongWritable(1);
        for (SegToken st: tokens) {
            if (st.word.matches("[\u4e00-\u9fa5]{2,}")) {  // 匹配中文词语
                context.write(new Text(st.word), one);
            }
        }
    }
}

get(false, "/output/cnwordcount/part-r-00000", "D:\\iso\\data\\cnword.txt", true);

/* File: Test4 */
package demo;

import ...

public class Test4 {
    public static void main(String[] args) {
        InputStream is = null;
        InputStreamReader isr = null;
        BufferedReader br = null;

        try {
            is = new FileInputStream("D:\\iso\\data\\cnword.txt");
            isr = new InputStreamReader(is, StandardCharsets.UTF_8);
            br = new BufferedReader(isr);
            List<Word> words = new ArrayList<>();
            String line = null;
            while ((line = br.readLine()) != null) {
                Word wd = new Word();
                String[] ss = line.split("\\s+");
                wd.key = ss[0];
                wd.value = Integer.parseInt(ss[1]);
                words.add(wd);
            }
            words.sort(new Comparator<Word>() {
                @Override
                public int compare(Word o1, Word o2) {
                    return o2.value - o1.value;  // 从大到小，降序排序
                }
            });
            for (int i = 0; i < words.size() && i < 10; i++) {  // 输出前 10 个词语
                System.out.println(words.get(i).key + " " + words.get(i).value);
            }
        } catch (Exception ex) {
        } finally {  // 关闭输入流
            try {
                br.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            try {
                isr.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            try {
                is.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }
}

/* File: ConcateText */
package cn.edu.nnu.homework;

import ...

public class ConcateText {

    private static Configuration cong = null;
    private static FileSystem fs = null;

    public static void main(String[] args) throws URISyntaxException, IOException {
        System.setProperty("HADOOP_USER_NAME", "lc");
        cong = new Configuration();
        fs = FileSystem.get(new URI("hdfs://master:8020"), cong);
        concat();
    }

    public static void concat() throws IOException {
        // 名著列表
        String[] files = {"hongloumeng-utf8.txt", "sanguoyanyi-utf8.txt",
              "shuihu-utf8.txt", "xiyouji-utf8.txt"};
        // 获取文件路径
        File path = new File("D:\\iso\\data");
        // 创建 hdfs 文件
        FSDataOutputStream out = fs.create(new Path("/test1127/masterpieces.txt"));
    // 输入流
        InputStream is = null;
        InputStreamReader isr = null;
        BufferedReader br = null;
        // 遍历四大名著
        for (String file : files) {
            try {
                is = new FileInputStream(new File(path, file));
                isr = new InputStreamReader(is, StandardCharsets.UTF_8);
                br = new BufferedReader(isr);
                String line = null;
                while ((line = br.readLine()) != null) {
                    if (line.trim().equals("")) continue;
                    while (true) {
                        // 取出最后一个字符
                        char lastChar = line.charAt(line.length() - 1);
                        // 如果是汉字，那么再读取一行，加到 line 上
                        if (lastChar >= 0x4e00 && lastChar <= 0x9fa5) {
                            String newline = br.readLine();
                            // 如果新的一行存在并且不为空
                            if (newline != null) {
                                if (!newline.trim().equals("")) {
                                    // 就拼上这一行
                                    line += newline.trim();
                                }
                            } else {
                                break;
                            }
                        } else {
                            break;  // 空则退出
                        }
                    }
          // 写入流
                    out.write(line.getBytes(StandardCharsets.UTF_8));
                }
            } catch (Exception ex) {
                ex.printStackTrace();
            } finally {
                try {
                    br.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
                try {
                    isr.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
                try {
                    is.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }
        // 关闭输出流
        out.close();
    }
}

package demo;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

public class Test4 {
    public static void main(String[] args) {
        InputStream is = null;
        InputStreamReader isr = null;
        BufferedReader br = null;

        try {
            is = new FileInputStream("D:\\iso\\data\\mastercount.txt");
            isr = new InputStreamReader(is, StandardCharsets.UTF_8);
            br = new BufferedReader(isr);
            List<Word> words = new ArrayList<>();
            String line = null;
            while ((line = br.readLine()) != null) {
                Word wd = new Word();
                String[] ss = line.split("\\s+");
                wd.key = ss[0];
                wd.value = Integer.parseInt(ss[1]);
                words.add(wd);
            }
            words.sort(new Comparator<Word>() {
                @Override
                public int compare(Word o1, Word o2) {
                    return o2.value - o1.value;  // 从大到小，降序排序
                }
            });
            for (int i = 0; i < words.size() && i < 20; i++) {
                System.out.println(words.get(i).key + "=" + words.get(i).value);
            }
        } catch (Exception ex) {
        } finally {
            try {
                br.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            try {
                isr.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            try {
                is.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }
}