基于MR实现ngram语言模型

在大数据的今天，世界上任何一台单机都无法处理大数据，无论cpu的计算能力或者内存的容量。必须采用分布式来实现多台单机的资源整合，来进行任务的处理，包括离线的批处理和在线的实时处理。

鉴于上次开会讲了语言模型的发展，从规则到后来的NNLM。本章的目的就是锻炼动手能力，在知道原理的基础上，通过采用MR范式，自己实现一个ngram语言模型。

首先通过maven来管理相关包的依赖。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.dingheng</groupId>
    <artifactId>nragmMR</artifactId>
    <version>1.0-SNAPSHOT</version>

    <packaging>jar</packaging>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-core</artifactId>
            <version>1.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.12</version>
        </dependency>
    </dependencies>
</project>

然后直接上代码：

1.首先是driver，作为程序的启动文件。

　　

package com.dingheng;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class Driver {

    public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {

        // inputDir
        // outputDir
        // NumOfGram
        // topK

        String inputDir = args[0];
        String outputDir = args[1];
        String numOfGram = args[2];
        String threshold = args[3];
        String topK = args[4];

        // first mapreduce
        Configuration configurationNGram = new Configuration();
        configurationNGram.set("textinputformat.recode.delimiter", ".");
        configurationNGram.set("numOfGram", numOfGram);

        Job jobNGram = Job.getInstance(configurationNGram);
        jobNGram.setJobName("NGram");
        jobNGram.setJarByClass(Driver.class);

        jobNGram.setMapperClass(NGram.NGramMapper.class);
        jobNGram.setReducerClass(NGram.NGramReducer.class);

        jobNGram.setOutputKeyClass(Text.class);
        jobNGram.setMapOutputValueClass(IntWritable.class);

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

        TextInputFormat.addInputPath(jobNGram, new Path(inputDir));
        TextOutputFormat.setOutputPath(jobNGram, new Path(outputDir));
        jobNGram.waitForCompletion(true);

        // second mapreduce
        Configuration configurationLanguage = new Configuration();
        configurationLanguage.set("threshold", threshold);
        configurationLanguage.set("topK", topK);

        DBConfiguration.configureDB(configurationLanguage,
                "com.mysql.jdbc.Driver",
                "jdbc:mysql://localhost:3306/test",
                "root",
                "123456");

        Job jobLanguage = Job.getInstance(configurationLanguage);
        jobLanguage.setJobName("LanguageModel");
        jobLanguage.setJarByClass(Driver.class);

        jobLanguage.setMapperClass(LanguageModel.Map.class);
        jobLanguage.setReducerClass(LanguageModel.Reduce.class);

        jobLanguage.setMapOutputKeyClass(Text.class);
        jobLanguage.setMapOutputValueClass(Text.class);
        jobLanguage.setOutputKeyClass(DBOutputWritable.class);
        jobLanguage.setOutputValueClass(NullWritable.class);

        jobLanguage.setInputFormatClass(TextInputFormat.class);
        jobLanguage.setOutputFormatClass(DBOutputFormat.class);

        DBOutputFormat.setOutput(
                jobLanguage,
                "output",
                new String[] { "starting_phrase", "following_word", "count"});

        TextInputFormat.setInputPaths(jobLanguage, new Path(args[1]));


        jobLanguage.waitForCompletion(true);

    }
}

2.然后是自己的定制类，自己定制了output

　

package com.dingheng;

import org.apache.hadoop.mapreduce.lib.db.DBWritable;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

public class DBOutputWritable implements DBWritable{

    private String starting_phrase;
    private String following_word;
    private int count;

    public DBOutputWritable(String starting_phrase, String following_word, int count) {
        this.starting_phrase = starting_phrase;
        this.following_word = following_word;
        this.count = count;
    }

    public void write(PreparedStatement arg0) throws SQLException {
        arg0.setString(1, starting_phrase);
        arg0.setString(2, following_word);
        arg0.setInt(3, count);
    }

    public void readFields(ResultSet arg0) throws SQLException {
        this.starting_phrase = arg0.getString(1);
        this.following_word = arg0.getString(2);
        this.count = arg0.getInt(3);
    }
}

3.之后自己的mapper和reducer。我试用了两个MR迭代，每一个迭代写在文件中

package com.dingheng;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class NGram {

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

        int numOfGram;

        @Override
        public void setup(Context context) {
            Configuration conf = context.getConfiguration();
            numOfGram = conf.getInt("numOfGram", 5);
        }

        @Override
        public void map(LongWritable key,
                        Text value,
                        Context context) throws IOException, InterruptedException {
            /*
            input: read sentence
            I love data n=3
            I love -> 1
            love data -> 1
            I love data -> 1
            */

            String line = value.toString().trim().toLowerCase().replaceAll("[^a-z]", " ");
            String[] words = line.split("\s+");

            if (words.length < 2) {
                return;
            }

            StringBuilder sb;
            for (int i = 0; i < words.length; i++) {
                sb = new StringBuilder();
                sb.append(words[i]);
                for (int j = 1; i + j < words.length && j < numOfGram; j++) {
                    sb.append(" ");
                    sb.append(words[i + j]);
                    context.write(new Text(sb.toString()), new IntWritable(1));
                }
            }
        }
    }

    public static class NGramReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        @Override
        public void reduce(Text key,
                           Iterable<IntWritable> values,
                           Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value: values) {
                sum = sum + value.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }
}

package com.dingheng;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;


public class LanguageModel {

    public static class Map extends Mapper<LongWritable, Text, Text, Text> {

        // input: I love big data	10
        // output: key: I love big  value: data = 10

        int threshold;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            Configuration configuration = context.getConfiguration();
            threshold = configuration.getInt("threshold", 20);
        }

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

            if ((value == null) || (value.toString().trim().length() == 0)) {
                return;
            }

            String line = value.toString().trim();

            String[] wordsPlusCount = line.split("	");
            String[] words = wordsPlusCount[0].split("\s+");
            int count = Integer.valueOf(wordsPlusCount[wordsPlusCount.length - 1]);

            if (wordsPlusCount.length < 2 || count < threshold) {
                return;
            }

            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < words.length - 1; i++) {
                sb.append(words[i]);
                sb.append(" ");
            }

            String outputKey = sb.toString().trim();
            String outputValue = words[words.length - 1];
            if (!(outputKey.length() < 1)) {
                context.write(new Text(outputKey), new Text(outputValue + "=" + count));
            }
        }
    }

    public static class Reduce extends Reducer<Text, Text, DBOutputWritable, NullWritable> {

        int topK;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            Configuration configuration = context.getConfiguration();
            topK = configuration.getInt("topK", 5);
        }

        @Override
        public void reduce(Text key,
                           Iterable<Text> values,
                           Context context) throws IOException, InterruptedException {
            // key: I love big
            // value: <data = 10, girl = 100, boy = 1000 ...>
            TreeMap<Integer, List<String>> tm = new TreeMap<Integer, List<String>>(Collections.<Integer>reverseOrder());
            // <10, <data, baby...>>, <100, <girl>>, <1000, <boy>>

            for (Text val : values) {
                // val: data = 10
                String value = val.toString().trim();
                String word = value.split("=")[0].trim();
                int count = Integer.parseInt(value.split("=")[1].trim());

                if (tm.containsKey(count)) {
                    tm.get(count).add(word);
                } else {
                    List<String> list = new ArrayList<String>();
                    list.add(word);
                    tm.put(count, list);
                }
            }

            Iterator<Integer> iter = tm.keySet().iterator();
            for (int j = 0; iter.hasNext() && j < topK; ) {
                int keyCount = iter.next();
                List<String> words = tm.get(keyCount);
                for (String curWord: words) {
                    context.write(new DBOutputWritable(key.toString(), curWord, keyCount), NullWritable.get());
                    j++;
                }
            }
        }
    }
}