1、HDFS文件操作
[命令行方式]
Hadoop的文件命令采取的形式为:
hadoop fs -cmd <args>
其中,cmd是具体的文件命令,而<args>是一组数目可变的参数。
(1)添加文件和目录
HDFS有一个默认的工作目录/user/$USER,其中$USER是你的登录用户名。不过这个目录不会自动建立,让我们用mkdir命令创建它。Hadoop的mkdir命令会自动创建父目录,类似于UNIX中使用-p选项的mkdir命令。
hadoop fs -mkdir /user/chuck
如果想看到所有的子目录,则可以使用hadoop的lsr命令,类似于UNIX中打开-r选项的ls:
hadoop fs -lsr /
[输出结果显示出属性信息,比如权限、所有者、组、文件大小以及最后修改日期,所有这些都类似于UNIX的概念。显示“1”的列给出文件的复制因子。因为复制因子不适用于目录,故届时该列仅会显示一个破折号(-)]
在本地文件系统中创建一个名为examle.txt的文本文件,用hadoop的put命令将它从本地文件系统复制到HDFS中:
hadoop fs -put example.txt ./
(2)获取文件
从HDFS中复制文件到本地文件系统:
hadoop fs -get example.txt ./
显示HDFS中文件的内容:
hadoop fs -cat example.txt
[可以在hadoop的文件命令中使用UNIX的管道,将其结果发送给其他的UNIX命令做进一步处理]
查看最后一千字节:
hadoop fs -tail example.txt
(3)删除文件
删除HDFS中的文件:
hadoop fs -rm example.txt
[ rm命令还可以用于删除空目录]
删除目录(目录不为空):
hadoop fs -rmr /user/chuck
(4)查阅帮助
hadoop fs -help <cmd>
[编程方式]
hadoop命令行工具中有一个getmerge命令,用于把一组HDFS文件在复制到本地计算机以前进行合并,下面开发的是实现把本地计算机文件复制到HDFS以前进行合并:
代码清单 PutMerge程序
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)根据用户定义的参数设置本地目录和HDFS的目标文件;
(2)提取本地输入目录中每个文件的信息;
(3)创建一个输出流写入到HDFS文件;
(4)遍历本地目录中的每个文件,打开一个输入流来读取该文件
FileSystem类还有些方法用于其他标准文件操作,如delete()、exists()、mkdirs()和rename()。
2、剖析MapReduce程序
MapReduce程序通过操作键/值对来处理数据,一般形式为:
map:(k1, v1) ——> list(k2, v2)
reduce:(k2, list(v2)) ——> list(k3,v3)
虽然我们可以并且的确经常把某些键与值称为整数、字符串等,但它们实际上并不是Integer、String等那些标准的Java类。这是因为为了让键/值对可以在集群上移动,MapReduce框架提供了一种序列化键/值对的方法。因此,只有那些支持这种序列化的类能够在这个框架中充当键或者值。
更具体而言,实现Writable接口的类可以是值,而实现WritableComparable<T>接口的类既可以是键也可以是值。注意WritableComparable<T>接口是Writable和java.lang.Comparable<T>接口的组合。对于键而言,我们需要这个比较,因为它们将在Reduce阶段进行排序,而值仅会被简单地传递。
Hadoop带有一些预定义的类用于实现WritableComparable,包括面向所有基本数据类型的封装类,如下表:
类 | 描述 |
BooleanWritable | 标准布尔变量的封装 |
ByteWritable | 单字节数的封装 |
DoubleWritable | 双字节数的封装 |
FloatWritable | 浮点数的封装 |
IntWritable | 整数的封装 |
LongWritable | 长整数的封装 |
Text | 使用UTF8格式的文本封装 |
NullWritable | 无键值的占位符 |
键和值所采用的数据类型可以超过Hadoop自身所支持的基本类型,可以自定义数据类型,只要它实现了Writable(或WritableComparable<T>)接口。
代码清单 示例实现WritableComparable接口的类
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 }
这个Edge类实现了Writable接口的readFields()及write()方法。它们与Java中的DataInput和DataOutput类一起用于类中内容的串行化。而Comparable接口中的实现是compareTo()方法。如果被调用的Edge小于、等于或者大于给定的Edge,这个方法会分别返回-1,0,1。
[Mapper]
一个类要作为mapper,需继承MapReducebase基类并实现Mapper接口。并不奇怪,mapper和reducer的基类均为MapReduceBase类。它包含类的构造与解构方法。
Mapper接口负责数据处理阶段。它采用的形式为Mapper<k1,v1,k2,v2>Java泛型,这里键类和值类分别实现WritableComparable和Writable接口。Mapper只有一个方法——Map,用于处理一个单独的键/值对。
void map (k1 key, v1 value, OutputCollector<k2,v2> output, Reporter reporter) throws IOException
该函数处理一个给定的键/值对 (k1,v1),生成一个键/值对(k2,v2)的列表(该列表也可能为空)。OutputCollector接收这个映射过程的输出,Reporter可以提供对mapper相关附加信息的记录,形成任务进度。
Hadoop提供了一些有用的mapper实现,如下表:
类 | 描述 |
IdentityMapper<k,v> | 实现Mapper<k,v,k,v>将输入直接映射到输出 |
InverseMapper<k,v> | 实现Mapper<k,v,v,k>反转键/值对 |
RegexMapper<k> | 实现Mapper<k,text,text,LongWritable>,为每个常规表达式的匹配项生成一个(match,1)对 |
TokenCountMapper<k> | 实现Mapper<k,text,text,LongWritable>,当输入的值为分词时,生成一个(token,1)对 |
[Reducer]
reducer的实现和mapper一样必须首先在MapReduce基类上扩展,允许配置和清理。此外,它还必须实现Reducer接口使其具有如下的单一方法:
void reduce(k2 key, Iterator<v2> values, OutputCollector<k3,v3> output, Reporter reporter) throws IOException
当reducer任务接收来自各个mapper的输出时,它按照键/值对中的键对输入数据进行排序,并将相同键的值归并。然后调用reduce()函数,并通过迭代处理那些与指定键相关联的值,生成一个(可能为空的)列表(k3,v3)。OutputCollector接收reduce阶段的输出,并写入输出文件。Reporter可提供对reducer相关附加信息的记录,形成任务进度。
Hadoop提供了一些基本的reducer实现,如下表:
类 | 描述 |
IdentityReudcer<k,v> | 实现Reducer<k,v,k,v>将输入直接映射到输出 |
LongSumReducer<k> | 实现<k,LongWritable,k,LongWritable>, 计算与给定键相对应的所有值的和 |
[Partitioner:重定向Mapper输出]
当使用多个reducer时,我们就需要采取一些办法来确定mapper应该把键/值对输出给谁。默认的作法是对键进行散列来确定reducer。hadoop通过HashPartitioner类强制执行这个策略。但有时HashPartitioner会让你出错。
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 }
一个定制的partitioner只需要实现configure()和getPartition()两个函数。前者将hadoop对作业的配置应用在patittioner上,而后者返回一个介于0和reducer任务数之间的整数,指向键/值对将要发送的reducer。
在map和reduce阶段之间,一个MapReduce应用必然从mapper任务得到输出结果,并把这些结果发布给reduce任务。该过程通常被称为洗牌。
[Combiner:本地reduce]
在许多MapReduce应用场景中,我们不妨在分发mapper结果之前做一下“本地Reduce”。
[预定义的mapper和reducer类的单词计数]
代码清单 修改的WordCount例程
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 }
3、读和写
[InputFormat]
hadoop分割与读取输入文件的方式被定义在InputFormat接口的一个实现中。TextInputFormat是InputFormat的默认实现,当你想要一次获取一行内容而输入数据又没有确定的键值时,这种数据格式通常会非常有用。
常用的InputFormat类,如下表:
InputFormat | 描述 |
TextInputFormat | 在文本文件中每一行均为一个记录。键(key)为一行的字节偏移,而值(value)为一行的内容 key: LongWritable value: Text |
KeyValueTextInputFormat | 在文本文件中的每一行均为一个记录。以每行的第一个分隔符为界,分隔符之前的是键(key),之后的是值(value)。分离器在属性key.value.separator.in.input.line中设定,默认为制表符(/t)。 key: Text Value: Text |
SequenceFileInputFormat<k,v> | 用于读取序列文件的InputFormat。键和值由用户定义。序列文件为hadoop专用的压缩二进制文件格式。它专用于一个MapReduce作业和其他MapReduce作业之间传送数据。 key: K(用户定义) value: V(用户定义) |
NLineInputFormat | 与TextInputFormat相同,但每个分片一定有N行。N在属性mapred.line.input.format.linespermap中设定,默认为1. key: LongWritable value: Text |
可以设置JobConf对象使用KeyValueTextInputFormat类读取这个文件:
conf.setInputFormat(KeyValueTextInputFormat.class);
回想一下,我们之前在mapper中曾使用LongWritable和Text分别作为键(key)和值(value)的类型。在TextInputFormat中,因为值为用数字表示的偏移量,所以LongWritable是一个合理的键类型。而当使用KeyvalueTextInputFormat时,无论是键和值都为Text类型,你必须改变mapper的实现以及map()方法来适应这个新的键(key)类型。
生成一个定制的InputFormat:略
[OutputFormat]
当MapReduce输出数据到文件时,使用的是OutputForamt类,它与inputForamt类相似。因为每个reducer仅需将它的输出写入自己的文件中,输出无需分片。输出文件放在一个公用目录中,通常命名为part-nnnnn,这里nnnnn是reducer的分区ID。RecordWriter对象将输出结果进行格式化,而RecordReader对输入格式进行解析。
常用的OutputFormat类,如下表:
OutputFormat | 描述 |
TextOutputFormat<k,v> | 将每个记录写为一行文本。键和值以字符串的形式写入,并以制表符(/t)分隔。这个分隔符可以在属性mapred.textoutputformat.separator中修改 |
SequenceFileOutputFormat<k,v> | 以hadoop专有序列文件格式写入键/值对。与SequenceFileInputForamt配合使用 |
NullOutputFormat<k,v> | 无输出 |