本地运行模式

  1. 在windows的eclipse里买能直接运行main方法,就会将job提交给本地执行其localjobrunner执行
    输入输出数据可以放在本度路径下(c:/file/src)
    输入输出数据可以也放在hdfs中(hdfs://muxiaocao:9000/file/src)
  2. 在linux打eclise里面直接运行main方法,但是不要添加yarn相关的配置,也会提交给localjobrunner执行
    输入输出数据可以放在本度路径下(/home/muxiaocao/file/src)
    输入输出数据也可以放在hdfs中(hdfs://muxiaocao:9000/file/src)

具体代码

mapper

package com.hadoop.muxiaocao.mapreduce;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MyMapper 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 line = value.toString();
        String[] words = StringUtils.split(line, " ");
        for (String word : words) {
            context.write(new Text(word),new LongWritable(1L) );
        }
    }


}

reducer

package com.hadoop.muxiaocao.mapreduce;

import java.io.IOException;

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

public class MyReducer 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 count = 0;
        for (LongWritable value : values) {
            count += value.get();
        }
        context.write(key, new LongWritable(count));
    }


}

jobrunner

package com.hadoop.muxiaocao.mapreduce;

import java.io.IOException;

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;

public class MyJobRunner {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 得到job对象
        Job myWcJob = Job.getInstance(new Configuration());
        // 设置可运行的class对象,hadoop利用这个类来查找包行它的jar文件,进而找到相关的jar文件
        myWcJob.setJarByClass(MyJobRunner.class);
        // 设置map 和 reducer的class对象,hadoop以此来找到map和reducer中的相关重写方法,实现本地逻辑
        myWcJob.setMapperClass(MyMapper.class);
        myWcJob.setReducerClass(MyReducer.class);
        // 设置输出的key和value类型
        myWcJob.setOutputKeyClass(Text.class);
        myWcJob.setOutputValueClass(LongWritable.class);

        // 设置输入输出文件。输出结果路径如果已经存在,hadoop就会报错并拒绝运行作业,以此来防止数据的丢失
        //FileInputFormat.addInputPath(myWcJob, new Path("/home/muxiaocao/myFile/srcfile/"));
        FileInputFormat.addInputPath(myWcJob, new Path("hdfs://muxiaocao:9000/"));
        FileOutputFormat.setOutputPath(myWcJob, new Path("/home/muxiaocao/myFile/output1/"));

        // 等待运行作业。hadoop yarn中的resourcemanager有一个作业队列,如果作业较多就会等待
        // 参数如果是true表示需要把运行过程打印到控制台
        myWcJob.waitForCompletion(true);
    }
}

集群运行模式

  1. 将工程打成jar包,上传到服务器,然后用hadoop命令提交hadoop jar wc.jar com.nanjing.muxiaocao.hadoop.wordcount
  2. 在linux的eclipse中,直接运行main方法,也可以提交到集群中运行,但是,必须采取一下措施:

    1. 在工程src目录下加入mapper-site.xml 和 yarn-site.xml配置文件
    2. 将工程打成jar包(wc.jar),同时在main方法中添加爱一个conf打配置参数
      conf.set("mapreduce.job.jar","wc.jar")
    3. 在windows的eclipse中直接诶运行main方法,也可以提交给集群中运行,但是因为平台不兼容,需要做很多设置

    要在windows中存放一份hadoop的安装包(解压好的)
    要将其中的lib和bin目录替换成根据你的windows版本重新编译出的文件
    再要配置系统环境变量 HADOOP_HOME 和 PATH
    修改YarnRunner这个类的源码

hadoop源码分析如何实现模式自动匹配

  • 对于本地和集群两种模式,最大的区别就在于集群模式中的main方法里,多了一个configuer配置信息,即
    conf.set("mapreduce.job.jar","wc.jar")
  • 另外集群模式下,也会配有mapper-site.xml 和 yarn-site.xml配置文件,hadoop源码中就是根据这些配置进行自动模式识别的。

关键源码如下

private void initialize(InetSocketAddress jobTrackAddr, Configuration conf)
      throws IOException {

    synchronized (frameworkLoader) {
      // 通过对frameworkLoader遍历,来分别看采用哪种模式
      // ClientProtocolProvider总共就两种LocalClientProtocolProvider和YarnClientProtocolProvider
      for (ClientProtocolProvider provider : frameworkLoader) {
        LOG.debug("Trying ClientProtocolProvider : "
            + provider.getClass().getName());
        ClientProtocol clientProtocol = null; 
        try {
          if (jobTrackAddr == null) {
            // 尝试创建模式
            clientProtocol = provider.create(conf);
          } else {
            clientProtocol = provider.create(jobTrackAddr, conf);
          }

          if (clientProtocol != null) {
            clientProtocolProvider = provider;
            client = clientProtocol;
            LOG.debug("Picked " + provider.getClass().getName()
                + " as the ClientProtocolProvider");
            break;
          }
          else {
            LOG.debug("Cannot pick " + provider.getClass().getName()
                + " as the ClientProtocolProvider - returned null protocol");
          }
        } 
        catch (Exception e) {
          LOG.info("Failed to use " + provider.getClass().getName()
              + " due to error: ", e);
        }
      }
    }

    if (null == clientProtocolProvider || null == client) {
      throw new IOException(
          "Cannot initialize Cluster. Please check your configuration for "
              + MRConfig.FRAMEWORK_NAME
              + " and the correspond server addresses.");
    }
  }

下面是两个ClientProtocolProvider的create方法

// LocalClientProtocolProvider的create方法
@Override
  public ClientProtocol create(Configuration conf) throws IOException {
    // 通过查看配置文件信息,来选择是否创建
    String framework =
        conf.get(MRConfig.FRAMEWORK_NAME, MRConfig.LOCAL_FRAMEWORK_NAME);
    if (!MRConfig.LOCAL_FRAMEWORK_NAME.equals(framework)) {
      return null;
    }
    conf.setInt(JobContext.NUM_MAPS, 1);

    return new LocalJobRunner(conf);
  }
  //YarnClientProtocolProvider的create方法
  @Override
  public ClientProtocol create(Configuration conf) throws IOException {
    if (MRConfig.YARN_FRAMEWORK_NAME.equals(conf.get(MRConfig.FRAMEWORK_NAME))) {
      return new YARNRunner(conf);
    }
    return null;
  }

注意:转载请说明,来自转自itboy-木小草尊重原创,尊重技术

Logo

更多推荐