模拟MapReduce编程的程序案例(用于统计文本中单词出现频率)

cn.toto.bigdata.mymr.task

TaskProcessor

核心的主体执行程序

ProcessLogic

定义客户端调用必须实现的方法，相当于WebService中的接口规范

cn.toto.bigdata.mymr.io

InputFormat

封装读文件的组件（接口用途）

DefaultInputFormat

封装读文件的组件的实现类

OutPutFormat

封装写文件的组件（接口用途）

DefaultOutPutFormat

封装写文件的组件的实现

cn.toto.bigdata.mymr.common

Constants

常量定义

Context

应用上下文，用于存储计算单词出现字数的次数的中间变量

cn.toto.bigdata.mymr.userapp

UserLogic

客户端对ProcessLogic规范的实现

UserApp

客户端主入口程序

cn.toto.bigdata.mymr.scheduler

Runner

客户端UserApp执行命令是依赖的Runner类，通过这里面的Socket发送命令。

WorkerClient

客户端执行时需要用到的client相关的代码

WorkerServer

UserApp执行时，需要提前启动的服务端

WorkerRunnable

服务端执行的相关逻辑

1、将mapreduce-my-demo导出成test.jar放置在E:/test.jar下。

2、需要有用于统计用的文本文件a.txt,文件在E:\a.txt

内容截图类似：

假设a.txt内容为：

The true

nobility is

in being

superior to

your previous

self guess

No great

discovery

was ever

made without

a bold

Knowledge will

give you

power but

character respect

The sun

is just

rising in

the morning

of another

day I

I figure

life is

a gift

and I

don’t intend

on wasting

3、首先运行：WorkerServer，相当于是启动服务端的代码

4、再次运行：UserApp，相当于是客户端

5、最终的统计结果将显示在：E:/out.txt中。统计结果如下：

nobility     1

but   1

gift   1

wasting    1

rising        1

don't         1

another    1

I        3

your 1

Knowledge       1

sun   1

without    1

life    1

The  2

character 1

and  1

of      1

power       1

just  1

day   1

you   1

on     1

No    1

a       2

give  1

figure        1

previous   1

in      2

will   1

made        1

was  1

is      3

being        1

bold 1

great        1

respect    1

morning   1

the   1

ever 1

superior   1

guess        1

discovery 1

true 1

self   1

to     1

intend       1

6、最终的日志将存储在：E:/task/task.log,最终的配置和工作用的jar也将会生成到这个目录下面，效果如下：

其中job.conf的内容为：

生成的task.log效果如下：

cn.toto.bigdata.mymr.task

TaskProcessor

核心的主体执行程序

ProcessLogic

定义客户端调用必须实现的方法，相当于WebService中的接口规范

TaskProcessor代码如下

package cn.toto.bigdata.mymr.task;

import java.util.HashMap;

import java.util.logging.FileHandler;

import java.util.logging.Level;

import java.util.logging.Logger;

import cn.toto.bigdata.mymr.common.Constants;

import cn.toto.bigdata.mymr.common.Context;

import cn.toto.bigdata.mymr.io.InputFormat;

import cn.toto.bigdata.mymr.io.OutPutFormat;

/**

 * 1、核心的主体执行程序

 * 这里是任务执行者

 */

public class TaskProcessor {

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

      // 加载用户指定的所有配置参数到上下文对象中，同时读取配置文件

      Context context = new Context();

      //获取上下文中的配置文件

      HashMap<String, String>  conf = context.getConfiguration();

      //通过打印日志的方式查看程序运行的结果

      Logger logger = Logger.getLogger("TaskProcessor");

      //设置日志的输出级别是INFO级别

      logger.setLevel(Level.INFO);

      FileHandler fileHandler = new FileHandler("E:/task/task.log");

      fileHandler.setLevel(Level.INFO);

      logger.addHandler(fileHandler);

      logger.info("context:" + context);

      logger.info("conf:" + conf);

      //初始化文件读取组件

      //从配置文件中获取用于读取的组件的class信息

      Class<?> forName = Class.forName(conf.get(Constants.INPUT_FORMAT));

      InputFormat inputFormat = (InputFormat) forName.newInstance();

      inputFormat.init(context);

      //用inputFormat组件读数据，并调用用户逻辑

      Class<?> forName2 = Class.forName(conf.get(Constants.USER_LOGIC));

      ProcessLogic userLogic = (ProcessLogic) forName2.newInstance();

      //对每一行调用用户逻辑，并通过context将用户调用结果存储内部缓存

      while(inputFormat.hasNext()) {

         Integer key = inputFormat.nextKey();

         String value = inputFormat.nextValue();

         userLogic.process(key, value, context);

      }

      userLogic.cleanUp(context);

      //替用户输出结果

      Class<?> forName3 = Class.forName(conf.get(Constants.OUTPUT_FORMAT));

      OutPutFormat outputFormat = (OutPutFormat) forName3.newInstance();

      outputFormat.write(context);

   }

}

ProcessLogic代码如下：

package cn.toto.bigdata.mymr.task;

import cn.toto.bigdata.mymr.common.Context;

/**

 * 1、规定的业务逻辑编写规范

 * process() 和  cleanUp都没有写实现，这里的实现在客户端

 */

public abstract class ProcessLogic {

   /**

    * 这里的context存储处理后的结果值

    * @param key          :行号

    * @param value        ：所在行的一行内容

    * @param context      :应用上下文的内容

    */

   public abstract void process(Integer key,String value,Context context);

   /**

    * 通过CleanUp输出处理后的结果

    */

   public void cleanUp(Context context){}

}

cn.toto.bigdata.mymr.io

InputFormat

封装读文件的组件（接口用途）

DefaultInputFormat

封装读文件的组件的实现类

OutPutFormat

封装写文件的组件（接口用途）

DefaultOutPutFormat

封装写文件的组件的实现

package cn.toto.bigdata.mymr.io;

import cn.toto.bigdata.mymr.common.Context;

public abstract class InputFormat {

   /**

    * 获取下一行要读的行的位置

    */

   public abstract int nextKey();

   /**

    * 获取从文件中读取的到的行的信息

    */

   public abstract String nextValue();

   /**

    * 从文件中读取到一行信息

    */

   public abstract String readLine() throws Exception;

   /**

    * 判断是否还可以读取到下一行的内容

    */

   public abstract boolean hasNext() throws Exception;

   /**

    * 初始化要读取的文件的路径和文件流

    */

   public abstract void init(Context context) throws Exception;

}

package cn.toto.bigdata.mymr.io;

import java.io.BufferedReader;

import java.io.FileReader;

import cn.toto.bigdata.mymr.common.Constants;

import cn.toto.bigdata.mymr.common.Context;

/**

 * 这里是默认的读取的实现类

 */

public class DefaultInputFormat extends InputFormat{

   //这里表示要读取的文件的路径

   private String inputPath;

   private BufferedReader br = null;

   //这里的key是指文本中类似读取到的指针的偏移量，是行号的偏移量

   private int key;

   //这里的value是指一行中的数据

   private String value;

   //默认读取的行是第0行

   private int lineNumber = 0;

   @Override

   public void init(Context context) throws Exception {

      //获取要读的文件的路径

       this.inputPath = context.getConfiguration().get(Constants.INPUT_PATH);

       //开始初始化输入流，只不过，这个流是从文件中获取的

       this.br = new BufferedReader(new FileReader(inputPath));

   }

   @Override

   public int nextKey() {

      return this.key;

   }

   @Override

   public String nextValue() {

      return this.value;

   }

   @Override

   public boolean hasNext() throws Exception {

      String line = null;

      line = readLine();

      //数据读取完成之后行号加一

      this.key = lineNumber++;

      this.value = line;

      return null != line;

   }

   /**

    * 读取一行数据

    */

   @Override

   public String readLine() throws Exception {

      String line = br.readLine();

      //如果读取到空了之后，将BufferedReader的值变成空

      if (line == null) {

         br.close();

      }

      return line;

   }

}

package cn.toto.bigdata.mymr.io;

import cn.toto.bigdata.mymr.common.Context;

/**

 * 用于输出结果的类

 */

public abstract class OutPutFormat {

   /**

    * 将结果写入文件中

    */

   public abstract void write(Context context) throws Exception;

   /**

    * 关闭流

    */

    public abstract void cleanUp() throws Exception;

}

package cn.toto.bigdata.mymr.io;

import java.io.BufferedWriter;

import java.io.FileWriter;

import java.util.HashMap;

import java.util.Set;

import java.util.Map.Entry;

import cn.toto.bigdata.mymr.common.Constants;

import cn.toto.bigdata.mymr.common.Context;

public class DefaultOutPutFormat extends OutPutFormat{

    BufferedWriter bw = null;

         @Override

         public void write(Context context) throws Exception {

             String outputPath = context.getConfiguration().get(Constants.OUTPUT_PATH);

             HashMap<String, Integer> KVBuffer = context.getKVBuffer();

             this.bw = new BufferedWriter(new FileWriter(outputPath));

             Set<Entry<String, Integer>> entrySet = KVBuffer.entrySet();

             for (Entry<String, Integer> entry : entrySet) {

                            bw.write(entry.getKey() + "\t" + entry.getValue() + "\r");

                   }

             bw.flush();

         }

         @Override

         public void cleanUp() throws Exception {

                   bw.close();

         }

}

cn.toto.bigdata.mymr.common

Constants

常量定义

Context

应用上下文，用于存储计算单词出现字数的次数的中间变量

package cn.toto.bigdata.mymr.common;

public class Constants {

   public static final String JAR_PATH = "jar.path";

   public static final String JAR_FILE = "job.jar";

   public static final String WORKER_HOST = "worker.host";

   public static final String WORKER_PORT = "worker.port";

   public static final String CONF_FILE = "job.conf";

   public static final String INPUT_FORMAT = "input.format.class";

   public static final String OUTPUT_FORMAT = "output.format.class";

   public static final String INPUT_PATH = "input.path";

   public static final String OUTPUT_PATH = "output.path";

   public static final String TASK_PROCESSOR = "cn.toto.bigdata.mymr.task.TaskProcessor";

   public static final String USER_LOGIC = "user.logic.class";

   public static final String TASK_WORK_DIR = "E:/task";

}

package cn.toto.bigdata.mymr.common;

import java.io.File;

import java.io.FileInputStream;

import java.io.ObjectInputStream;

import java.util.HashMap;

/**

 * 应用上下文，通过这个内容获取配置文件

 * 通过这个上下文最终输出结果

 */

public class Context {

         private HashMap<String, Integer> KVBuffer = new HashMap<String, Integer>();

         private HashMap<String, String> conf;

         @SuppressWarnings("unchecked")

         public Context() throws Exception {

                   //加载配置参数

                   File file = new File(Constants.TASK_WORK_DIR + "/" + Constants.CONF_FILE);

                   if (file.exists()) {

                            @SuppressWarnings("resource")

                            ObjectInputStream oi = new ObjectInputStream(new FileInputStream(file));

                            this.conf = (HashMap<String, String>) oi.readObject();

                   } else {

                            // throw new RuntimeException("read conf failed ....");

                   }

         }

         /**

          * 通过这种变量最后输出结果

          */

         public void write(String k, Integer v) {

                   KVBuffer.put(k, v);

         }

         public HashMap<String, Integer> getKVBuffer() {

                   return KVBuffer;

         }

         public void setKVBuffer(HashMap<String, Integer> tmpKV) {

                   this.KVBuffer = tmpKV;

         }

         /**

          * 获取配置文件中的信息

          */

         public HashMap<String, String> getConfiguration() {

                   return conf;

         }

         /**

          * 在Context()构造函数里面已经有了conf的配置，这里再次传入说明配置可以让用户手动指定

          */

         public void setConfiguration(HashMap<String, String> configuration) {

                   this.conf = configuration;

         }

}

cn.toto.bigdata.mymr.userapp

UserLogic

客户端对ProcessLogic规范的实现

UserApp

客户端主入口程序

package cn.toto.bigdata.mymr.userapp;

import java.util.HashMap;

import java.util.Set;

import java.util.Map.Entry;

import cn.toto.bigdata.mymr.common.Context;

import cn.toto.bigdata.mymr.task.ProcessLogic;

public class UserLogic extends ProcessLogic {

         private HashMap<String, Integer> wordCount = new HashMap<String, Integer>();

         @Override

         public void process(Integer key, String value, Context context) {

                   String [] words = value.split(" ");

                   for(String word : words) {

                            Integer count = wordCount.get(word);

                            if (count == null) {

                                     wordCount.put(word, 1);

                            } else {

                                     wordCount.put(word, count + 1);

                            }

                   }

         }

         public void cleanUp(Context context) {

                   Set<Entry<String, Integer>> entrySet = wordCount.entrySet();

                   for(Entry<String, Integer> entry : entrySet) {

                            context.write(entry.getKey(), entry.getValue());

                   }

         }

}

package cn.toto.bigdata.mymr.userapp;

import java.util.HashMap;

import cn.toto.bigdata.mymr.common.Constants;

import cn.toto.bigdata.mymr.scheduler.Runner;

public class UserApp {

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

                   HashMap<String, String> conf = new HashMap<String,String>();

                   conf.put(Constants.INPUT_PATH, "E:/a.txt");

                   conf.put(Constants.OUTPUT_PATH, "E:/out.txt");

                   conf.put(Constants.INPUT_FORMAT, "cn.toto.bigdata.mymr.io.DefaultInputFormat");

                   conf.put(Constants.OUTPUT_FORMAT, "cn.toto.bigdata.mymr.io.DefaultOutPutFormat");

                   conf.put(Constants.JAR_PATH, "E:/test.jar");

                   conf.put(Constants.WORKER_HOST, "localhost");

                   conf.put(Constants.WORKER_PORT, "9889");

                   conf.put(Constants.USER_LOGIC, "cn.toto.bigdata.mymr.userapp.UserLogic");

                   Runner runner = new Runner(conf);

                   runner.submit("localhost", 9889);

         }

}

cn.toto.bigdata.mymr.scheduler

Runner

客户端UserApp执行命令是依赖的Runner类，通过这里面的Socket发送命令。

WorkerClient

客户端执行时需要用到的client相关的代码

WorkerServer

UserApp执行时，需要提前启动的服务端

WorkerRunnable

服务端执行的相关逻辑

package cn.toto.bigdata.mymr.scheduler;

import java.io.FileOutputStream;

import java.io.ObjectOutputStream;

import java.util.HashMap;

import cn.toto.bigdata.mymr.common.Constants;

public class Runner {

    private HashMap<String, String> conf;

    public Runner(HashMap<String, String> conf) {

             this.conf = conf;

    }

    public void submit(String host,int port) throws Exception {

             ObjectOutputStream jobConfStream = new ObjectOutputStream(new FileOutputStream(Constants.CONF_FILE));

                   jobConfStream.writeObject(conf);

                   WorkerClient workerClient = new WorkerClient(conf);

                   workerClient.submit();

    }

}

package cn.toto.bigdata.mymr.scheduler;

import java.io.FileInputStream;

import java.io.OutputStream;

import java.net.Socket;

import java.util.HashMap;

import cn.toto.bigdata.mymr.common.Constants;

public class WorkerClient {

         private HashMap<String, String> conf;

         Socket socket = null;

         OutputStream so = null;

         public WorkerClient(HashMap<String, String> conf) {

                   this.conf = conf;

         }

         public void submit() throws Exception {

                   socket = new Socket(conf.get(Constants.WORKER_HOST), Integer.parseInt(conf.get(Constants.WORKER_PORT)));

                   so = socket.getOutputStream();

                   String jarPath = conf.get(Constants.JAR_PATH);

                   // 发送jar包

                   byte[] buff = new byte[4096];

                   FileInputStream jarIns = new FileInputStream(jarPath);

                   so.write("jarfile".getBytes());

                   int read = 0;

                   while ((read=jarIns.read(buff)) != -1) {

                            so.write(buff,0,read);

                   }

                   jarIns.close();

                   so.close();

                   socket.close();

                   // 发送job.conf文件

                   socket = new Socket(conf.get(Constants.WORKER_HOST), Integer.parseInt(conf.get(Constants.WORKER_PORT)));

                   so = socket.getOutputStream();

                   FileInputStream confIns = new FileInputStream(Constants.CONF_FILE);

                   so.write("jobconf".getBytes());

                   while ((read = confIns.read(buff)) != -1) {

                            so.write(buff,0,read);

                   }

                   confIns.close();

                   so.close();

                   socket.close();

                   // 发送启动命令

                   socket = new Socket(conf.get(Constants.WORKER_HOST), Integer.parseInt(conf.get(Constants.WORKER_PORT)));

                   so = socket.getOutputStream();

                   so.write("job2run".getBytes());

                   String shell = "java -cp E:/test.jar cn.toto.bigdata.mymr.task.TaskProcessor";

                   so.write(shell.getBytes());

                   so.close();

                   socket.close();

         }

}

package cn.toto.bigdata.mymr.scheduler;

import java.net.ServerSocket;

import java.net.Socket;

public class WorkerServer {

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

                   ServerSocket ssc = new ServerSocket(9889);

                   System.out.println("Worker服务器启动-->9889");

                   while (true) {

                            Socket accept = ssc.accept();

                            new Thread(new WorkerRunnable(accept)).start();

                   }

         }

}

package cn.toto.bigdata.mymr.scheduler;

import java.io.BufferedReader;

import java.io.File;

import java.io.FileOutputStream;

import java.io.InputStream;

import java.io.InputStreamReader;

import java.net.Socket;

import cn.toto.bigdata.mymr.common.Constants;

public class WorkerRunnable implements Runnable {

         Socket socket;

         InputStream in = null;

         volatile long confSize = 0;

         volatile long jarSize = 0;

         public WorkerRunnable(Socket socket) {

                   this.socket = socket;

         }

         @Override

         public void run() {

                   try {

                            this.in = socket.getInputStream();

                            byte[] protocal = new byte[7];

                            int read = in.read(protocal, 0, 7);

                            if (read < 7) {

                                     System.out.println("客户端请求不符合协议规范......");

                                     return;

                            }

                            String command = new String(protocal);

                            switch (command) {

                            case "jarfile":

                                     receiveJarFile();

                                     break;

                            case "jobconf":

                                     receiveConfFile();

                                     break;

                            case "job2run":

                                     runJob();

                                     break;

                            default:

                                     System.out.println("客户端请求不符合协议规范.....");

                                     socket.close();

                                     break;

                            }

                   } catch (Exception e) {

                            e.printStackTrace();

                   }

         }

         private void receiveConfFile() throws Exception {

                   System.out.println("开始接收conf文件");

                   FileOutputStream fo = new FileOutputStream(Constants.TASK_WORK_DIR + "/" + Constants.CONF_FILE);

                   byte[] buff = new byte[4096];

                   int read = 0;

                   while ((read = in.read(buff)) != -1) {

                            confSize += read;

                            fo.write(buff, 0, read);

                   }

                   fo.flush();

                   fo.close();

                   in.close();

                   socket.close();

         }

         private void receiveJarFile() throws Exception {

                   System.out.println("开始接收jar文件");

                   FileOutputStream fo = new FileOutputStream(Constants.TASK_WORK_DIR + "/" + Constants.JAR_FILE);

                   byte[] buff = new byte[4096];

                   int read = 0;

                   while ((read = in.read(buff)) != -1) {

                            jarSize += read;

                            fo.write(buff, 0, read);

                   }

                   fo.flush();

                   fo.close();

                   in.close();

                   socket.close();

         }

         private void runJob() throws Exception {

                   byte[] buff = new byte[4096];

                   int read = in.read(buff);

                   String shell = new String(buff, 0, read);

                   System.out.println("接收到启动命令......." + shell);

                   in.close();

                   socket.close();

                   Thread.sleep(500);

                   File jarFile = new File(Constants.TASK_WORK_DIR + "/" + Constants.JAR_FILE);

                   File confFile = new File(Constants.TASK_WORK_DIR + "/" + Constants.CONF_FILE);

                   System.out.println("jarfile 存在？" + jarFile.exists());

                   System.out.println("confFile 存在？" + confFile.exists());

                   System.out.println("jarfile可读？" + jarFile.canRead());

                   System.out.println("jarfile可写？" + jarFile.canWrite());

                   System.out.println("confFile可读？" + confFile.canRead());

                   System.out.println("confFile可写？" + confFile.canWrite());

                   System.out.println("jarFile.length():" + jarFile.length());

                   System.out.println("confFile.length():" + confFile.length());

                   /*if (jarFile.length() == jarSize && confFile.length() == confSize) {

                            System.out.println("jar 和 conf 文件已经准备就绪......");

                   }*/

                   System.out.println("开始启动数据处理TaskProcessor......");

                   Process exec = Runtime.getRuntime().exec(shell);

                   int waitFor = exec.waitFor();

                   InputStream errStream = exec.getErrorStream();

                   BufferedReader errReader = new BufferedReader(new InputStreamReader(errStream));

                   String inLine = null;

                   /*

                    * InputStream stdStream = exec.getInputStream(); BufferedReader

                    * stdReader = new BufferedReader(new InputStreamReader(stdStream));

                    * while ((inLine = stdReader.readLine()) != null) {

                    * System.out.println(inLine); }

                    */

                   while ((inLine = errReader.readLine()) != null) {

                            System.out.println(inLine);

                   }

                   if (waitFor == 0) {

                            System.out.println("task成功运行完毕.....");

                   } else {

                            System.out.println("task异常退出......");

                   }

         }

}