我尝试在CentOS中构建一个Hadoop Mapreduce程序来检查输入文件上的列。文件只包含文本，不包含XML，看起来像这样:
Apple|Orange|Grape|Apple
Banana|Apple|Melon
Melon|Orange
Apple|Banana|Grape
Melon|Orange
用分隔符'|'分隔列。我的程序旨在检查每行上的列数By附加每一列的第一个字符，如
苹果|橙|葡萄|苹果-> AOGA
键建立后，程序将计算每个键的长度以检查每行中有多少列。实际上，我将使用这些键来区分列数超过指定限制的行。输出数据格式为:
(键、Keys.length Rowcount)
我的预期结果是:
AOGA 4 1
BAM 3 1
ABG 31 1
MO 2 2
这是我的源代码: Columncheck.java
package com.mapreduce;
import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class Columncheck {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = new Job(conf, "Columncheck");
//Set class which run from jar file
job.setJarByClass(Columncheck.class);
//Set Key class datatype
job.setOutputKeyClass(Text.class);
//Set summary output datatype
job.setOutputValueClass(CountTuple.class);
//Set Mapper and Reducer class
job.setMapperClass(Map.class);
job.setCombinerClass(Reduce.class);
job.setReducerClass(Reduce.class);
//Set input-output data format
// job.setInputFormatClass(TextInputFormat.class);
// job.setOutputFormatClass(TextOutputFormat.class);
//Declare Input and Output Path from Arguments (from Terminal)
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true)?0:1);
}
// Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
public static class Map extends Mapper<Object, Text, Text, CountTuple> {
private Text word = new Text(); //Value
private CountTuple outTuple = new CountTuple();
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString(); //A|B|C
StringTokenizer tokenizer = new StringTokenizer(line);
ArrayList<String> stringList = new ArrayList<String>();
ArrayList<String> stringList2 = new ArrayList<String>();
while (tokenizer.hasMoreTokens()) {
stringList.add(tokenizer.nextToken());
}
for(String item: stringList){ // item format => A|B|C
StringTokenizer tokenizer2 = new StringTokenizer(item,"|");
String tokens = "";
while (tokenizer2.hasMoreTokens()) {
tokens = tokens + tokenizer2.nextToken().charAt(0);
}
stringList2.add(tokens); //Output : ABC
}
for(String item2: stringList2){
outTuple.setLength(item2.length());
outTuple.setCount(1);
word.set(item2);
context.write(word, outTuple);
//System.out.println(outTuple.getLength()+ " " + outTuple.getCount());
}
//End of mapping
}
}
// Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> *********************
public static class Reduce extends Reducer<Text, CountTuple, Text, CountTuple> {
private CountTuple result = new CountTuple();
//Automatic shuffle keys
// This method is called at once for each key
public void reduce(Text key, Iterable<CountTuple> values, Context context)
throws IOException, InterruptedException {
result.setLength(0);
result.setCount(0);
int sum = 0;
int wordlength = 0;
for (CountTuple val : values) {
sum += val.getCount();
wordlength = val.getLength();
}
result.setLength(wordlength);
result.setCount(sum);
context.write(key, result);
}
}
}
和我的类:CountTuple.java
package com.mapreduce;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Writable;
public class CountTuple implements Writable{
private Integer wlength;
private long count;
public CountTuple() {
this.wlength = 0;
this.count = 0;
}
public Integer getLength() {
return wlength;
}
public void setLength(Integer i) {
this.wlength = i;
}
public long getCount() {
return count;
}
public void setCount(long count) {
this.count = count;
}
public void readFields(DataInput in) throws IOException {
wlength = in.readInt();
count = in.readLong();
}
public void write(DataOutput out) throws IOException {
out.writeInt(wlength);
out.writeLong(count);
}
}
这是来自控制台的消息:
-bash-4.1$ hadoop jar Columncheck.jar com.mapreduce.Columncheck /tmp/gphdtmp/colchkinput /tmp/gphdtmp/colchkoutput
14/08/19 19:00:23 INFO service.AbstractService: Service:org.apache.hadoop.yarn.client.YarnClientImpl is inited.
14/08/19 19:00:23 INFO service.AbstractService: Service:org.apache.hadoop.yarn.client.YarnClientImpl is started.
14/08/19 19:00:24 WARN mapreduce.JobSubmitter: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.
14/08/19 19:00:24 INFO input.FileInputFormat: Total input paths to process : 1
14/08/19 19:00:25 INFO mapreduce.JobSubmitter: number of splits:1
In DefaultPathResolver.java. Path = hdfs://hdname01:8020/tmp/gphdtmp/colchkoutput
14/08/19 19:00:25 WARN conf.Configuration: mapred.jar is deprecated. Instead, use mapreduce.job.jar
14/08/19 19:00:25 WARN conf.Configuration: mapred.output.value.class is deprecated. Instead, use mapreduce.job.output.value.class
14/08/19 19:00:25 WARN conf.Configuration: mapreduce.combine.class is deprecated. Instead, use mapreduce.job.combine.class
14/08/19 19:00:25 WARN conf.Configuration: mapreduce.map.class is deprecated. Instead, use mapreduce.job.map.class
14/08/19 19:00:25 WARN conf.Configuration: mapred.job.name is deprecated. Instead, use mapreduce.job.name
14/08/19 19:00:25 WARN conf.Configuration: mapreduce.reduce.class is deprecated. Instead, use mapreduce.job.reduce.class
14/08/19 19:00:25 WARN conf.Configuration: mapred.input.dir is deprecated. Instead, use mapreduce.input.fileinputformat.inputdir
14/08/19 19:00:25 WARN conf.Configuration: mapred.output.dir is deprecated. Instead, use mapreduce.output.fileoutputformat.outputdir
14/08/19 19:00:25 WARN conf.Configuration: mapred.map.tasks is deprecated. Instead, use mapreduce.job.maps
14/08/19 19:00:25 WARN conf.Configuration: mapred.output.key.class is deprecated. Instead, use mapreduce.job.output.key.class
14/08/19 19:00:25 WARN conf.Configuration: mapred.working.dir is deprecated. Instead, use mapreduce.job.working.dir
14/08/19 19:00:25 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1408091977394_0024
14/08/19 19:00:26 INFO client.YarnClientImpl: Submitted application application_1408091977394_0024 to ResourceManager at hdname00/10.14.233.41:8032
14/08/19 19:00:26 INFO mapreduce.Job: The url to track the job: http://hdname00-1:8088/proxy/application_1408091977394_0024/
14/08/19 19:00:26 INFO mapreduce.Job: Running job: job_1408091977394_0024
14/08/19 19:00:37 INFO mapreduce.Job: Job job_1408091977394_0024 running in uber mode : false
14/08/19 19:00:37 INFO mapreduce.Job: map 0% reduce 0%
14/08/19 19:00:46 INFO mapreduce.Job: map 100% reduce 0%
14/08/19 19:00:54 INFO mapreduce.Job: map 100% reduce 100%
14/08/19 19:00:54 INFO mapreduce.Job: Job job_1408091977394_0024 completed successfully
14/08/19 19:00:55 INFO mapreduce.Job: Counters: 43
File System Counters
FILE: Number of bytes read=78
FILE: Number of bytes written=175951
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
HDFS: Number of bytes read=207
HDFS: Number of bytes written=152
HDFS: Number of read operations=6
HDFS: Number of large read operations=0
HDFS: Number of write operations=2
Job Counters
Launched map tasks=1
Launched reduce tasks=1
Rack-local map tasks=1
Total time spent by all maps in occupied slots (ms)=14840
Total time spent by all reduces in occupied slots (ms)=20685
Map-Reduce Framework
Map input records=5
Map output records=5
Map output bytes=79
Map output materialized bytes=78
Input split bytes=115
Combine input records=5
Combine output records=4
Reduce input groups=4
Reduce shuffle bytes=78
Reduce input records=4
Reduce output records=4
Spilled Records=8
Shuffled Maps =1
Failed Shuffles=0
Merged Map outputs=1
GC time elapsed (ms)=59
CPU time spent (ms)=5030
Physical memory (bytes) snapshot=1075609600
Virtual memory (bytes) snapshot=6045433856
Total committed heap usage (bytes)=2024800256
Shuffle Errors
BAD_ID=0
CONNECTION=0
IO_ERROR=0
WRONG_LENGTH=0
WRONG_MAP=0
WRONG_REDUCE=0
File Input Format Counters
Bytes Read=92
File Output Format Counters
Bytes Written=152
运行源代码时没有出现错误，但是结果变成:
$ hadoop fs -cat /tmp/gphdtmp/colchkoutput/part-r-00000
ABG com.mapreduce.CountTuple@2cee0cd1
AOGA com.mapreduce.CountTuple@2cee0cd1
BAM com.mapreduce.CountTuple@2cee0cd1
MO com.mapreduce.CountTuple@2cee0cd1
您需要在您的CountTuple中覆盖#toString方法。
例如:
@Override
public String toString() {
return count + "";
}