簡單描述一下我的問題，我想用MapReducec處理資料，Map函式寫好能跑出來，但是reduce函式一直出錯，由于Map函式比較復雜，這里只給出Map函式的輸出，以及我的Reduce函式，希望各位大神們看到可能幫我解答一下~Hadoop初學者，身邊沒有非常懂的同學~~自己在家狂試，但是還是沒解決~謝謝大家了~
Map輸出：
key value
112008 張峰計算機資訊工程 1
計算機 資訊工程 1
181678 沈玉日語外國語 2
日語 外國語 2
112009 張峰計算機經濟管理 3
計算機 經濟管理 3
（其中value的值中間以空格分割的）
Reduce處理：
將相同的Key值合并，這里只有兩個計算機的key值相同，其他的都自成一個key，遍歷同一個key的value，
如 計算機 資訊工程 1
經濟管理 3
按照空格分隔每一個valuelist，如將資訊工程賦值給split[0]將1賦值給split[1],然后直接將其輸出，接下來同樣的，將經濟管理賦值給split[0]將3賦值給split[1],然后直接將其輸出。原始程式后面還有步驟但是后來發現這一部就不對了。。所以想請大家幫忙看看~
我的Reduce 部分如下：
public static class YCLReducer extends Reducer< Text, Text,Text,Text> {
//private LongWritable result = new LongWritable();
private Text a1 = new Text();
private Text b1 = new Text();
public void reduce(Text key, Iterable<Text> values,
Context context
) throws IOException, InterruptedException {
String a;
String b;
Iterator ite=values.iterator();
while(ite.hasNext())
{
String record=ite.next().toString();
String [] lineSplit = record.split(" ");
a1.set(lineSplit[0]);
b1.set(lineSplit[1]);
context.write(a1,b1);
}
/*
for(int m=0;m<ziduan.size();m++)
{
jlh1.set(jiluhao.get(m));
jlh2.set(jiluhao.get(m));
context.write(jlh1,jlh2);
}
*/
/*for(int n=m+1;n<ziduan.size();n++)
{
if(ziduan.get(m)!=ziduan.get(n))
{
jlh1.set(jiluhao.get(m));
jlh2.set(jiluhao.get(n));
context.write(jlh1,jlh2);
}
}
*/
}
}
詳細代碼如下：
資料：1 112008 zhangfeng jisuanji xinxigongcheng
import java.io.IOException;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.util.GenericOptionsParser;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;
public class YCLyilai extends Configured implements Tool
{
enum Counter
{
LINESKIP,//chu cuo de hang
}
public static class YCLMapper extends Mapper<LongWritable, Text, Text, Text>
{
// private final static LongWritable one = new LongWritable(1);
private Text zhuma = new Text();
private Text out = new Text();
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
{
String linevalue = value.toString();
String [] relateSplit="1 2 3 4 p 3 4 p".split(" ");
StringTokenizer stringTokenizer = new StringTokenizer(linevalue);
try
{
String [] lineSplit = linevalue.split(" ");
String t=null;
String s=" ";
for(int m=0;m<relateSplit.length;m++)
{
if(!relateSplit[m].equals("p"))
{
s+=lineSplit[Integer.parseInt(relateSplit[m])]+" ";
}
else
{
String [] partSplit = s.split(" ");
zhuma.set(partSplit[1]);
String s1=" ";
for(int j=2;j<partSplit.length;j++)
{
s1+=partSplit[j];
}
out.set(s1+" "+lineSplit[0]);
context.write(zhuma,out);
s=" ";
}
}
}
catch(java.lang.ArrayIndexOutOfBoundsException e)
{
context.getCounter(Counter.LINESKIP).increment(1);
return;
}
}
}
public static class YCLReducer extends Reducer< Text, Text,Text,Text> {
//private LongWritable result = new LongWritable();
private Text a1 = new Text();
private Text b1 = new Text();
public void reduce(Text key, Iterable<Text> values,
Context context
) throws IOException, InterruptedException {
String a;
String b;
Iterator ite=values.iterator();
while(ite.hasNext())
{
String record=ite.next().toString();
String [] lineSplit = record.split(" ");
a1.set(lineSplit[0]);
b1.set(lineSplit[1]);
context.write(a1,b1);
}
/*
for(int m=0;m<ziduan.size();m++)
{
jlh1.set(jiluhao.get(m));
jlh2.set(jiluhao.get(m));
context.write(jlh1,jlh2);
}
*/
/*for(int n=m+1;n<ziduan.size();n++)
{
if(ziduan.get(m)!=ziduan.get(n))
{
jlh1.set(jiluhao.get(m));
jlh2.set(jiluhao.get(n));
context.write(jlh1,jlh2);
}
}
*/
}
}
public int run(String[] args)throws Exception
{
Configuration conf=getConf();
Job job=new Job(conf,"YCLyilai");
job.setJarByClass(YCLyilai.class);
FileInputFormat.addInputPath(job, new Path("hdfs://master:9000/DSYCLyilai_input2"));
FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/DSYCLyilai_output3"));
// job.setJarByClass(YCLyilai.class);
job.setMapperClass(YCLMapper.class);
job.setCombinerClass(YCLReducer.class);
job.setReducerClass(YCLReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.waitForCompletion(true);
return job.isSuccessful()?0:1;
}
public static void main(String[] args) throws Exception {
int res=ToolRunner.run(new Configuration(), new YCLyilai(), args);
System.exit(res);
}
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/80936.html
標籤：云存儲
上一篇：求助 kibana
下一篇：spark streaming 中使用saveAsNewAPIHadoopDataset方法寫入hbase中，從checkpoint中恢復時報錯