MapReduce编程之倒排索引-白红宇

MapReduce编程之倒排索引

阅读量：6237 次

发布时间：2019-06-22

本文共 3644 字，大约阅读时间需要 12 分钟。

任务要求：

//输入文件格式

18661629496 110

13107702446 110

1234567 120

2345678 120

987654 110

2897839274 18661629496

//输出文件格式格式

11018661629496|13107702446|987654|18661629496|13107702446|987654|

1201234567|2345678|1234567|2345678|

186616294962897839274|2897839274|

mapreduce程序编写：

                            
import
java.io.IOException;
import
java.util.StringTokenizer;
import
org.apache.hadoop.conf.Configuration;
import
org.apache.hadoop.fs.Path;
import
org.apache.hadoop.io.IntWritable;
import
org.apache.hadoop.io.LongWritable;
import
org.apache.hadoop.io.Text;
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;
public
class
Test2 { 
    
enum
Counter
    
{ 
        
LINESKIP,
//记录出错的行
    
}
    
public
static
class
Map 
extends
Mapper<LongWritable, Text, Text, Text>{ 
        
public
void
map(LongWritable key, Text value, Context context)
                
throws
IOException, InterruptedException { 
             
String line = value.toString();
//读取源数据
             
try
             
{ 
                 
//数据处理
                 
String [] lineSplit = line.split(
" "
);
//18661629496,110
                 
String anum = lineSplit[
0
];
                 
String bnum = lineSplit[
1
];  
                 
//输出格式：110,18661629496               
                 
context.write(
new
Text(bnum), 
new
Text(anum));
                 
             
}
             
catch
(ArrayIndexOutOfBoundsException e)
             
{ 
                 
context.getCounter(Counter.LINESKIP).increment(
1
);
//出错时计数器+1
                 
return
;
             
}
        
}
    
}
    
public
static
class
Reduce 
extends
Reducer<Text, Text, Text, Text> { 
        
public
void
reduce(Text key, Iterable<Text> values, Context context)
                
throws
IOException, InterruptedException { 
            
String valueString;
            
String out=
""
;
            
for
(Text value:values)
            
{ 
                
valueString=value.toString();
                
out+=valueString+
"|"
;
            
}
            
context.write(key, 
new
Text(out));
        
}
    
}
    
public
static
void
main(String[] args) 
throws
Exception { 
        
Configuration conf = 
new
Configuration();
        
if
(args.length != 
2
) { 
            
System.err.println(
"请配置输入输出路径 "
);
            
System.exit(
2
);
        
}
        
//各种配置
        
Job job = 
new
Job(conf, 
"telephone "
);
//作业名称配置
        
//类配置
        
job.setJarByClass(Test2.
class
);
        
job.setMapperClass(Map.
class
);
        
job.setReducerClass(Reduce.
class
);
        
//map输出格式配置
        
job.setMapOutputKeyClass(Text.
class
);
        
job.setMapOutputValueClass(Text.
class
);
        
//作业输出格式配置
        
job.setOutputKeyClass(Text.
class
);
        
job.setOutputValueClass(Text.
class
);
        
//添加输入输出路径
        
FileInputFormat.addInputPath(job, 
new
Path(args[
0
]));
        
FileOutputFormat.setOutputPath(job, 
new
Path(args[
1
]));
        
//任务完毕时退出
        
System.exit(job.waitForCompletion(
true
) ? 
0
: 
1
);
    
}
}
         
       