
需要通过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;
}
}
数据分析咨询请扫描二维码
若不方便扫码,搜微信号:CDAshujufenxi
CDA 数据分析师:就业前景广阔的新兴职业 在当今数字化时代,数据已成为企业和组织决策的重要依据。数据分析师作为负责收集 ...
2025-06-30探秘卷积层:为何一个卷积层需要两个卷积核 在深度学习的世界里,卷积神经网络(CNN)凭借其强大的特征提取能力 ...
2025-06-30探索 CDA 数据分析师在线课程:开启数据洞察之旅 在数字化浪潮席卷全球的当下,数据已成为企业决策、创新与发展的核心驱 ...
2025-06-303D VLA新范式!CVPR冠军方案BridgeVLA,真机性能提升32% 编辑:LRST 【新智元导读】中科院自动化所提出BridgeVLA模型,通过将 ...
2025-06-30LSTM 为何会产生误差?深入剖析其背后的原因 在深度学习领域,LSTM(Long Short-Term Memory)网络凭借其独特的记忆单元设 ...
2025-06-27LLM进入拖拽时代!只靠Prompt几秒定制大模型,效率飙升12000倍 【新智元导读】最近,来自NUS、UT Austin等机构的研究人员创新 ...
2025-06-27探秘 z-score:数据分析中的标准化利器 在数据的海洋中,面对形态各异、尺度不同的数据,如何找到一个通用的标准来衡量数据 ...
2025-06-26Excel 中为不同柱形设置独立背景(按数据分区)的方法详解 在数据分析与可视化呈现过程中,Excel 柱形图是展示数据的常用工 ...
2025-06-26CDA 数据分析师会被 AI 取代吗? 在当今数字化时代,数据的重要性日益凸显,数据分析师成为了众多企业不可或缺的角色 ...
2025-06-26CDA 数据分析师证书考取全攻略 在数字化浪潮汹涌的当下,数据已成为企业乃至整个社会发展的核心驱动力。数据分析师作 ...
2025-06-25人工智能在数据分析的应用场景 在数字化浪潮席卷全球的当下,数据以前所未有的速度增长,传统的数据分析方法逐渐难以满足海 ...
2025-06-25评估模型预测为正时的准确性 在机器学习与数据科学领域,模型预测的准确性是衡量其性能优劣的核心指标。尤其是当模型预测结 ...
2025-06-25CDA认证:数据时代的职业通行证 当海通证券的交易大厅里闪烁的屏幕实时跳动着市场数据,当苏州银行的数字金融部连夜部署新的风控 ...
2025-06-24金融行业的大数据变革:五大应用案例深度解析 在数字化浪潮中,金融行业正经历着深刻的变革,大数据技术的广泛应用 ...
2025-06-24Power Query 中实现移动加权平均的详细指南 在数据分析和处理中,移动加权平均是一种非常有用的计算方法,它能够根据不同数据 ...
2025-06-24数据驱动营销革命:解析数据分析在网络营销中的核心作用 在数字经济蓬勃发展的当下,网络营销已成为企业触达消费者 ...
2025-06-23随机森林模型与 OPLS-DA 的优缺点深度剖析 在数据分析与机器学习领域,随机森林模型与 OPLS-DA(正交偏最小二乘法判 ...
2025-06-23CDA 一级:开启数据分析师职业大门的钥匙 在数字化浪潮席卷全球的今天,数据已成为企业发展和决策的核心驱动力,数据分析师 ...
2025-06-23透视表内计算两个字段乘积的实用指南 在数据处理与分析的过程中,透视表凭借其强大的数据汇总和整理能力,成为了众多数据工 ...
2025-06-20CDA 一级考试备考时长全解析,助你高效备考 CDA(Certified Data Analyst)一级认证考试,作为数据分析师领域的重要资格认证, ...
2025-06-20