[hadoop in Action] 第3章 Hadoop组件

 1 import java.io.IOException;
 2 
 3 import org.apache.hadoop.conf.Configuration;
 4 import org.apache.hadoop.fs.FSDataInputStream;
 5 import org.apache.hadoop.fs.FSDataOutputStream;
 6 import org.apache.hadoop.fs.FileStatus;
 7 import org.apache.hadoop.fs.FileSystem;
 8 import org.apache.hadoop.fs.Path;
 9 
10 public class PutMerge {
11 
12     public static void main(String[] args) throws IOException {
13 
14         Configuration conf = new Configuration();
15         FileSystem hdfs  = FileSystem.get(conf);
16         FileSystem local = FileSystem.getLocal(conf);
17 
18         Path inputDir = new Path(args[0]);   //（1）设定输入目录和输出文件
19         Path hdfsFile = new Path(args[1]); 
20 
21         try {
22             FileStatus[] inputFiles = local.listStatus(inputDir);    //（2）得到本地文件列表
23             FSDataOutputStream out = hdfs.create(hdfsFile);    //（3）生成HDFS输出流
24 
25             for (int i=0; i<inputFiles.length; i++) {
26                 System.out.println(inputFiles[i].getPath().getName());
27                 FSDataInputStream in = local.open(inputFiles[i].getPath());    //（4）打开本地输入流
28                 byte buffer[] = new byte[256];
29                 int bytesRead = 0;
30                 while( (bytesRead = in.read(buffer)) > 0) {
31                     out.write(buffer, 0, bytesRead);
32                 }
33                 in.close();
34             }
35             out.close();
36         } catch (IOException e) {
37             e.printStackTrace();
38         }
39     }
40 }

 1 import java.io.DataInput;
 2 import java.io.DataOutput;
 3 import java.io.IOException;
 4  
 5 import org.apache.hadoop.io.WritableComparable;
 6  
 7 public class Edge implements WritableComparable<Edge> {
 8  
 9     private String departureNode;
10     private String arrivalNode;
11  
12     public String getDepartureNode() { return departureNode;}
13  
14     @Override
15     public void readFields(DataInput in) throws IOException {    //（1）说明如何读入数据
16         departureNode = in.readUTF();
17         arrivalNode = in.readUTF();     
18     }
19  
20     @Override
21     public void write(DataOutput out) throws IOException {    //（2）说明如何写入数据
22         out.writeUTF(departureNode);
23         out.writeUTF(arrivalNode); 
24     }
25  
26     @Override
27     public int compareTo(Edge o) {    //（3）定义数据排序
28      return (departureNode.compareTo(o.departureNode) != 0)
29          ? departureNode.compareTo(o.departureNode)
30          : arrivalNode.compareTo(o.arrivalNode);
31     }
32 }

 1 public class EdgePartitioner implements Partitioner<Edge, Writable>
 2 {
 3      @verride
 4      public int getPartition(Edge key, Writable value, int numPartitions)
 5      {
 6           return key.getDepartureNode().hashCode() % numPartitions;
 7      }
 8  
 9      @verride
10      public void configure(JobConf conf) { }
11 }

 1 import org.apache.hadoop.fs.Path;
 2 import org.apache.hadoop.io.Text;
 3 import org.apache.hadoop.io.LongWritable;
 4 import org.apache.hadoop.mapred.FileInputFormat;
 5 import org.apache.hadoop.mapred.FileOutputFormat;
 6 import org.apache.hadoop.mapred.JobClient;
 7 import org.apache.hadoop.mapred.JobConf;
 8 import org.apache.hadoop.mapred.lib.TokenCountMapper;
 9 import org.apache.hadoop.mapred.lib.LongSumReducer;
10  
11 public class WordCount2 {
12     public static void main(String[] args) {
13         JobClient client = new JobClient();
14         JobConf conf = new JobConf(WordCount2.class);
15  
16         FileInputFormat.addInputPath(conf, new Path(args[0]));
17         FileOutputFormat.setOutputPath(conf, new Path(args[1]));
18  
19         conf.setOutputKeyClass(Text.class);
20         conf.setOutputValueClass(LongWritable.class);
21         conf.setMapperClass(TokenCountMapper.class);
22         conf.setCombinerClass(LongSumReducer.class);
23         conf.setReducerClass(LongSumReducer.class);
24  
25         client.setConf(conf);
26         try {
27             JobClient.runJob(conf);
28         } catch (Exception e) {
29             e.printStackTrace();
30         }
31     }
32 }