你好,游客 登录 注册 发布搜索
背景:
阅读新闻

如何通过Java程序提交yarn的MapReduce计算任务

[日期:2014-11-14] 来源:Linux社区  作者: [字体: ]

  由于项目需求,需要通过Java程序提交Yarn的MapReduce的计算任务。与一般的通过Jar包提交MapReduce任务不同,通过程序提交MapReduce任务需要有点小变动,详见以下代码。

  以下为MapReduce主程序,有几点需要提一下:

  1、在程序中,我将文件读入格式设定为WholeFileInputFormat,即不对文件进行切分。

  2、为了控制reduce的处理过程,map的输出键的格式为组合键格式。与常规的<key,value>不同,这里变为了<textpair,value>,TextPair的格式为<key1,key2>。

  3、为了适应组合键,重新设定了分组函数,即GroupComparator。分组规则为,只要TextPair中的key1相同(不要求key2相同),则数据被分配到一个reduce容器中。这样,当相同key1的数据进入reduce容器后,key2起到了一个数据标识的作用。

  package web.Hadoop;

  import java.io.IOException;

  import org.apache.hadoop.conf.Configuration;

  import org.apache.hadoop.fs.Path;

  import org.apache.hadoop.io.BytesWritable;

  import org.apache.hadoop.io.WritableComparable;

  import org.apache.hadoop.io.WritableComparator;

  import org.apache.hadoop.mapred.JobClient;

  import org.apache.hadoop.mapred.JobConf;

  import org.apache.hadoop.mapred.JobStatus;

  import org.apache.hadoop.mapreduce.Job;

  import org.apache.hadoop.mapreduce.Partitioner;

  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

  import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;

  import util.Utils;

  public class GEMIMain {

  public GEMIMain(){

  job = null;

  }

  public Job job;

  public static class NamePartitioner extends

  Partitioner<textpair, byteswritable=""> {

  @Override

  public int getPartition(TextPair key, BytesWritable value,

  int numPartitions) {

  return Math.abs(key.getFirst().hashCode() * 127) % numPartitions;

  }

  }

  /**

  * 分组设置类,只要两个TextPair的第一个key相同,他们就属于同一组。他们的Value就放到一个Value迭代器中,

  * 然后进入Reducer的reduce方法中。

  *

  * @author hduser

  *

  */

  public static class GroupComparator extends WritableComparator {

  public GroupComparator() {

  super(TextPair.class, true);

  }

  @Override

  public int compare(WritableComparable a, WritableComparable b) {

  TextPair t1 = (TextPair) a;

  TextPair t2 = (TextPair) b;

  // 比较相同则返回0,比较不同则返回-1

  return t1.getFirst().compareTo(t2.getFirst()); // 只要是第一个字段相同的就分成为同一组

  }

  }

  public boolean runJob(String[] args) throws IOException,

  ClassNotFoundException, InterruptedException {

  Configuration conf = new Configuration();

  // 在conf中设置outputath变量,以在reduce函数中可以获取到该参数的值

  conf.set("outputPath", args[args.length - 1].toString());

  //设置HDFS中,每次任务生成产品的质量文件所在文件夹。args数组的倒数第二个原数为质量文件所在文件夹

  conf.set("qualityFolder", args[args.length - 2].toString());

  //如果在Server中运行,则需要获取web项目的根路径;如果以java应用方式调试,则读取/opt/hadoop-2.5.0/etc/hadoop/目录下的配置文件

  //MapReduceProgress mprogress = new MapReduceProgress();

  //String rootPath= mprogress.rootPath;

  String rootPath="/opt/hadoop-2.5.0/etc/hadoop/";

  conf.addResource(new Path(rootPath+"yarn-site.xml"));

  conf.addResource(new Path(rootPath+"core-site.xml"));

  conf.addResource(new Path(rootPath+"hdfs-site.xml"));

  conf.addResource(new Path(rootPath+"mapred-site.xml"));

  this.job = new Job(conf);

  job.setJobName("Job name:" + args[0]);

  job.setJarByClass(GEMIMain.class);

  job.setMapperClass(GEMIMapper.class);

  job.setMapOutputKeyClass(TextPair.class);

  job.setMapOutputValueClass(BytesWritable.class);

  // 设置partition

  job.setPartitionerClass(NamePartitioner.class);

  // 在分区之后按照指定的条件分组

  job.setGroupingComparatorClass(GroupComparator.class);

  job.setReducerClass(GEMIReducer.class);

  job.setInputFormatClass(WholeFileInputFormat.class);

  job.setOutputFormatClass(NullOutputFormat.class);

  // job.setOutputKeyClass(NullWritable.class);

  // job.setOutputValueClass(Text.class);

  job.setNumReduceTasks(8);

  // 设置计算输入数据的路径

  for (int i = 1; i < args.length - 2; i++) {

  FileInputFormat.addInputPath(job, new Path(args[i]));

  }

  // args数组的最后一个元素为输出路径

  FileOutputFormat.setOutputPath(job, new Path(args[args.length - 1]));

  boolean flag = job.waitForCompletion(true);

  return flag;

  }

  @SuppressWarnings("static-access")

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

  IOException, InterruptedException {

  String[] inputPaths = new String[] { "normalizeJob",

  "hdfs://192.168.168.101:9000/user/hduser/red1/",

  "hdfs://192.168.168.101:9000/user/hduser/nir1/","quality11111",

  "hdfs://192.168.168.101:9000/user/hduser/test" };

  GEMIMain test = new GEMIMain();

  boolean result = test.runJob(inputPaths);

  }

  }

  以下为TextPair类

  public class TextPair implements WritableComparable {

  private Text first;

  private Text second;

  public TextPair() {

  set(new Text(), new Text());

  }

  public TextPair(String first, String second) {

  set(new Text(first), new Text(second));

  }

  public TextPair(Text first, Text second) {

  set(first, second);

  }

  public void set(Text first, Text second) {

  this.first = first;

  this.second = second;

  }

  public Text getFirst() {

  return first;

  }

  public Text getSecond() {

  return second;

  }

  @Override

  public void write(DataOutput out) throws IOException {

  first.write(out);

  second.write(out);

  }

  @Override

  public void readFields(DataInput in) throws IOException {

  first.readFields(in);

  second.readFields(in);

  }

  @Override

  public int hashCode() {

  return first.hashCode() * 163 + second.hashCode();

  }

  @Override

  public boolean equals(Object o) {

  if (o instanceof TextPair) {

  TextPair tp = (TextPair) o;

  return first.equals(tp.first) && second.equals(tp.second);

  }

  return false;

  }

  @Override

  public String toString() {

  return first + "\t" + second;

  }

  @Override

  /**A.compareTo(B)

  * 如果比较相同,则比较结果为0

  * 如果A大于B,则比较结果为1

  * 如果A小于B,则比较结果为-1

  *

  */

  public int compareTo(TextPair tp) {

  int cmp = first.compareTo(tp.first);

  if (cmp != 0) {

  return cmp;

  }

  //此时实现的是升序排列

  return second.compareTo(tp.second);

  }

  }

  以下为WholeFileInputFormat,其控制数据在mapreduce过程中不被切分

  package web.hadoop;

  import java.io.IOException;

  import org.apache.hadoop.fs.Path;

  import org.apache.hadoop.io.BytesWritable;

  import org.apache.hadoop.io.Text;

  import org.apache.hadoop.mapreduce.InputSplit;

  import org.apache.hadoop.mapreduce.JobContext;

  import org.apache.hadoop.mapreduce.RecordReader;

  import org.apache.hadoop.mapreduce.TaskAttemptContext;

  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

  public class WholeFileInputFormat extends FileInputFormat<text, byteswritable=""> {

  @Override

  public RecordReader<text, byteswritable=""> createRecordReader(

  InputSplit arg0, TaskAttemptContext arg1) throws IOException,

  InterruptedException {

  // TODO Auto-generated method stub

  return new WholeFileRecordReader();

  }

  @Override

  protected boolean isSplitable(JobContext context, Path filename) {

  // TODO Auto-generated method stub

  return false;

  }

  }

  以下为WholeFileRecordReader类

  package web.hadoop;

  import java.io.IOException;

  import org.apache.hadoop.conf.Configuration;

  import org.apache.hadoop.fs.FSDataInputStream;

  import org.apache.hadoop.fs.FileSystem;

  import org.apache.hadoop.fs.Path;

  import org.apache.hadoop.io.BytesWritable;

  import org.apache.hadoop.io.IOUtils;

  import org.apache.hadoop.io.Text;

  import org.apache.hadoop.mapreduce.InputSplit;

  import org.apache.hadoop.mapreduce.RecordReader;

  import org.apache.hadoop.mapreduce.TaskAttemptContext;

  import org.apache.hadoop.mapreduce.lib.input.FileSplit;

  public class WholeFileRecordReader extends RecordReader<text, byteswritable=""> {

  private FileSplit fileSplit;

  private FSDataInputStream fis;

  private Text key = null;

  private BytesWritable value = null;

  private boolean processed = false;

  @Override

  public void close() throws IOException {

  // TODO Auto-generated method stub

  // fis.close();

  }

  @Override

  public Text getCurrentKey() throws IOException, InterruptedException {

  // TODO Auto-generated method stub

  return this.key;

  }

  @Override

  public BytesWritable getCurrentValue() throws IOException,

  InterruptedException {

  // TODO Auto-generated method stub

  return this.value;

  }

  @Override

  public void initialize(InputSplit inputSplit, TaskAttemptContext tacontext)

  throws IOException, InterruptedException {

  fileSplit = (FileSplit) inputSplit;

  Configuration job = tacontext.getConfiguration();

  Path file = fileSplit.getPath();

  FileSystem fs = file.getFileSystem(job);

  fis = fs.open(file);

  }

  @Override

  public boolean nextKeyValue() {

  if (key == null) {

  key = new Text();

  }

  if (value == null) {

  value = new BytesWritable();

  }

  if (!processed) {

  byte[] content = new byte[(int) fileSplit.getLength()];

  Path file = fileSplit.getPath();

  System.out.println(file.getName());

  key.set(file.getName());

  try {

  IOUtils.readFully(fis, content, 0, content.length);

  // value.set(content, 0, content.length);

  value.set(new BytesWritable(content));

  } catch (IOException e) {

  // TODO Auto-generated catch block

  e.printStackTrace();

  } finally {

  IOUtils.closeStream(fis);

  }

  processed = true;

  return true;

  }

  return false;

  }

  @Override

  public float getProgress() throws IOException, InterruptedException {

  // TODO Auto-generated method stub

  return processed ? fileSplit.getLength() : 0;

  }

  }

收藏 推荐 打印 | 录入:574107552 | 阅读:
相关新闻       MapReduce Java Yarn 
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款