Hadoop学习---Hadoop的深入学习

Hadoop生态圈

存储数据HDFS(Hadoop Distributed File System),运行在通用硬件上的分布式文件系统。具有高度容错性、高吞吐量的的特点。

处理数据MapReduce，它是一种编程模型，Map（映射）和Reduce（归约），它极大地方便了分布式并行编程，与hdfs的高度融合，它是基于java来进行编程的。

数据仓库工具Hive，处理结构化SQL查询功能，将sql语句解释为MapReduce编程进行数据的处理，只能进行结构化的查询。

Pig MapReduce之上的高级过程语言，查询大型的半结构化数据集，处理非结构化数据。

Hadoop的优势

低成本基于来源软件，运行与通用硬件平台。

高扩展性在可用的计算机集簇间分配数据并完成计算任务，这些集簇可以方便的扩展到数以千计的节点中。

高效性在节点之间动态分配计算任务并保证各个节点的动态平衡，处理速度非常快。

高容错性自动保存数据的多个副本，并且能够自动将失败的任务重新分配。

Hadoop三大核心设计

MapReduce Map:任务分解 Reduce：结果汇总

HDFS NameNode：管理里文件元数据 DataNode：存储物理文件 Client：获取文件的各种API

Hbase 列式内存数据库,它是一个适用于非结构化数据存储的数据库，基于列存储，不同行可有不同数据列，保留数据多个时间版本。

另： Zookeeper 分布式应用程序协调服务提供一致性服务。

HDFS的体系结构

NameNode：即元数据节点，用来管理文件系统的命名空间，记录每个文件数据在各个DataNode上的位置和副本信息，协调客户端对文件的访问。

元数据文件：VERSION、seen_txid、fsimage_*、fsimage_*.md5、edits_*

DataNode：即数据节点用来存储数据，负责所在物理节点的存储管理，一次写入，多次读取（不修改）。

文件由数据块组成，典型的块大小是64MB，数据块尽量散布到各个节点。

SecondaryNameNode：NameNode的一个快照，周期性的备份namenode，记录namenode中的metadata及其他数据，可以用来恢复Namenode

dfs.namenode.name.dir/current/文件夹的几个文件：

VERSION：存放的是版本信息，文件系统的标示符

seen_txid：存放事务ID的文件，它里面的数字要与edits的尾数一致，也就是与edits_inprogress_000000000*后边的数字一致，inprogress是正在使用的edits文件

fsimage_*、fsimage_*.md5、edits_*：存放的是文件的元数据信息。

HDFS文件读取的过程

HDFS文件读取的过程如下：

1).使用HDFS提供的客户端开发库Client，向远程的Namenode发起RPC请求；

2).Namenode会视情况返回文件的部分或者全部block列表，对于每个block，Namenode都会返回有该block拷贝的DataNode地址；

3).客户端开发库Client会选取离客户端最接近的DataNode来读取block；如果客户端本身就是DataNode,那么将从本地直接获取数据.

4).读取完当前block的数据后，关闭与当前的DataNode连接，并为读取下一个block寻找最佳的DataNode；

5).当读完列表的block后，且文件读取还没有结束，客户端开发库会继续向Namenode获取下一批的block列表。

6).读取完一个block都会进行checksum验证，如果读取datanode时出现错误，客户端会通知Namenode，然后再从下一个拥有该block拷贝的datanode继续读。

HDFS写入文件的过程

HDFS写入文件的过程：

1).使用HDFS提供的客户端开发库Client，向远程的Namenode发起RPC请求；

2).Namenode会检查要创建的文件是否已经存在，创建者是否有权限进行操作，成功则会为文件创建一个记录，否则会让客户端抛出异常；

4).当客户端开始写入文件的时候，开发库会将文件切分成多个packets，并在内部以数据队列"data queue"的形式管理这些packets，并向Namenode申请新的blocks，获取用来存储replicas的合适的datanodes列表，列表的大小根据在Namenode中对replication的设置而定。

开始以pipeline（管道）的形式将packet写入所有的replicas中。开发库把packet以流的方式写入第一个datanode，该datanode把该packet存储之后，再将其传递给在此pipeline中的下一个datanode，直到最后一个datanode，这种写数据的方式呈流水线的形式。

5).最后一个datanode成功存储之后会返回一个ack packet，在pipeline里传递至客户端，在客户端的开发库内部维护着"ack queue"，成功收到datanode返回的ack packet后会从"ack queue"移除相应的packet。

6).如果传输过程中，有某个datanode出现了故障，那么当前的pipeline会被关闭，出现故障的datanode会从当前的pipeline中移除，剩余的block会继续剩下的datanode中继续以pipeline的形式传输，同时Namenode会分配一个新的datanode，保持replicas设定的数量。

HDFS的shell操作

1. 直接访问Hadoop程序

在/etc/prifile中加入

export HADOOP_HOME=/home/hduser/hadoop （hadoop的安装目录）
export PATH=$HADOOP_HOME/bin:$PATH

source /etc/profile 是环境变量生效

2. HDFS命令格式：

HDFS基本命令（在hadoop目录下执行为例）

bin/hadoop fs -cmd <args>

           cmd:具体的操作，基本上与UNIX的命令行相同

           <args>:有时需包含参数

             例如：bin/hadoop fs -ls /

3. 常见的hadoop命令解析：

put命令只能从本地的文件复制到HDFS上；

get命令将HDFS上的文件复制到本地；

cp命令只能在相同的文件系统上互相复制文件。这三个命令都可以复制多个文件。复制单个文件时，目标路径可以是目录也可以是文件，目标路径是目录时，文件名不改变，目标路径是文件时，可以修改文件名；复制多个文件时，目标路径必须是目录，文件名不能修改。

copyFromLocal命令，此命令与put命令相似，区别是此命令只能复制一个文件。目标路径是目录时，文件名不改变，目标路径是文件时，可以修改文件名。

copyToLocal命令，此命令与get命令相似，区别是此命令只能复制一个文件。目标路径是目录时，文件名不改变，目标路径是文件时，可以修改文件名。

mv命令只能在相同的文件系统上移动文件，可以移动多个文件。标路径是目录时，文件名不改变，目标路径是文件时，可以修改文件名。

以上所有目录必须是存在的。

4. 常见的hadoop命令示例

mkdir 使用方法：hadoop fs -mkdir <paths>

      示例：hadoop fs -mkdir /user

ls 列出path目录下的内容，包括文件名，权限，所有者，大小和修改时间

　　hadoop fs -ls /

　　hadoop fs -ls -R /

put 使用方法：hadoop fs -put <localsrc> ... <dst>

　　从本地文件系统中复制单个或多个源路径到目标文件系统。也支持从标准输入中读取输入写入目标文件系统。

　　hadoop fs -put localfile /user/ 拷贝localfile文件到hdfs的user目录下

　　hadoop fs -put localfile1 localfile2 /user/ 同时拷贝localfile1和localfile2文件到hdfs的user目录下

　　hadoop fs -put - /user/hadoopfile 在hadoopfile文件里手工录入内容（录入之前hadoopfile文件不存在），按Ctrl+C键录入结束　

get 使用方法：hadoop fs -get [-ignorecrc] [-crc]    <src> <localdst> 复制文件到本地文件系统

　　示例：hadoop fs -get /user/hadoop/file localfile

cat 使用方法：hadoop fs -cat URI [URI ...]

　　示例：hadoop fs -ls /user/hadoopfile 查看文件内容

rm 使用方法：hadoop fs -rm URI [URI ...]

　　删除指定的文件。只删除非空目录和文件。请参考rmr命令了解递归删除。

　　示例：hadoop fs -rm /user/read.txt 删除文件

　　示例：hadoop fs -rm  -f /user/read.txt 强制删除文件

rmr 使用方法：hadoop fs -rmr URI [URI ...]

　　示例：hadoop fs -rmr /user/     删除/usr/下所有的文件和目录

delete的递归版本。

　　示例：hadoop fs -rm -r /user/hadoop/dir

getmerge 使用方法：hadoop fs -getmerge <src> <localdst> [addnl]

　　接受一个源目录和一个目标文件作为输入，并且将源目录中所有的文件连接成本地目标文件。addnl是可选的，用于指定在每个文件结尾添加一个换行符。

　　示例：hadoop fs -getmerge /input2 file2.txt

mv 使用方法：hadoop fs -mv URI [URI ...] <dest>

　　将文件从源路径移动到目标路径。这个命令允许有多个源路径，此时目标路径必须是一个目录。

　　不允许在不同的文件系统间移动文件。

　　也可以重命名文件

　　示例：hadoop fs -mv /hadoop/file1 /hadoop/file2

　　　　  hadoop fs -mv /hadoop/file1.txt  /hadoop/file1.txt.bak.0722　　

stat 使用方法：hadoop fs -stat URI [URI ...]

　　返回指定路径的统计信息。

　　示例：hadoop fs -stat /input2

tail 使用方法：hadoop fs -tail [-f] URI

　　将文件尾部1k字节的呢绒输出到stdout。支持-f选项，行为和Unix中一致。

　　示例：hadoop fs -tail pathname

chmod 使用方法：hadoop fs -chmod [-R] <MODE[,MODE]... | OCTALMODE> URI [URI ...]

　　设为所有人皆可读取

　　chmod a+r file1.txt

chown 使用方法：hadoop fs -chown [-R] [OWNER][:[GROUP]] URI [URI]

touchz 使用方法：hadoop fs -touchz URI [URI ...]

　　创建一个0字节的空文件。

　　示例：hadoop fs -touchz pathname

copyToLocal 使用方法：hadoop fs -copyToLocal [ignorecrc] [-crc] URI <localdst>

　　除了限定目标路径是一个本地文件外，和get命令类似。

copyFromLocal 使用方法：hadoop fs -copyFromLocal <localsrc> URI

　　除了限定源路径是一个本地文件外，和put命令相似。

cp 使用方法：hadoop fs -cp URI [URI ...] <dest>

　　将文件从源路径复制到目标路径。这个命令允许有多个源路径，此时目标路径必须是一个目录。

　　示例： hadoop fs -cp /user/hadoop/file1 /user/hadoop/file2

du 使用方法：hadoop fs -du URI [URI ...]

　　显示目录中所有文件的大小

　　示例：hadoop fs -du /user/hadoop/dir1

du -s 使用方法：hadoop fs -du -s <args>

　　显示文件的大小

　　常用：hadoop fs -du -h /srv/smart/hhh.txt

expunge 使用方法：hadoop fs -expunge
　　清空回收站
回收站： hadoop fs -ls /安装目录下/.trash 
       Hadoop文件删除后可以放在回收站内

5. HDFS的API方式访问

Eclipse中hadoop环境的建立

Configuration类：该类的对象封装了客户端或者服务端的配置。
FileSystem类：该类的对象是一个文件系统对象，可以用该队想的一些方法来对文件进行操作。
FSDataInputStream和FSDataOutputStream：这两个类是HDFS中的输入输出流。

基本流程：
得到Configuration对象
得到FileSystem对象
进行文件操作（读写、删除、改名）
所需引入的库

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

文件操作前的准备工作

获取配置信息
Configuration conf = new Configuration();
建立文件系统对象
FileSystem fs = FileSystem.get(conf);
写文件
Path dstPath = new Path(dst);//目标路径
FSDataOutputStream outputStream = fs.create(dstPath);//打开一个输出流
outputStream.write(contents);
outputStream.close();
fs.close();
读文件
InputStream in = null;
in = fs.open(srcPath);
IOUtils.copyBytes(in,System.out,4096,false);//复制到标准输出流
IOUtils.closeStream(in);
上传文件到HDFS
Path srcPath = new Path(src);//原路径
Path dstPath = new Path(dst);//目标路径
fs.copyFromLocalFile(false,srcPath,dstPath);
前面参数是指是否删除原文件，true为删除，默认为false
获取指定文件信息
FileStatus[] fileStatus = fs.listStatus(dstPath);
boolean isok=fs.rename(oldPath,newPath);
boolean isok=fs.deleteOnExit(path);
boolean isok=fs.mkdirs(srcPath);

6. HDFS API具体操作

  1 打开eclipse建立一个工程，将代码copy，逐步运行。
  2 
  3 在工程中建立一个conf目录，注意先将相关的配置文件拷贝过去主要是三个文件core-site.xml , mapred-site.xml,hdfs-site.xml
  4 
  5 并在project属性中设置java build path 单击addfolder 将conf目录加上
  6 
  7 import java.io.FileInputStream;
  8 
  9 import java.io.IOException;
 10 
 11 import java.io.InputStream;
 12 
 13 import org.apache.hadoop.conf.Configuration;
 14 
 15 import org.apache.hadoop.fs.FSDataOutputStream;
 16 
 17 import org.apache.hadoop.fs.FileStatus;
 18 
 19 import org.apache.hadoop.fs.FileSystem;
 20 
 21 import org.apache.hadoop.fs.Path;
 22 
 23 import org.apache.hadoop.io.IOUtils;
 24 
 25 public class HdfsFile {
 26 
 27     //创建新文件
 28 
 29     public static void createFile(String dst , byte[] contents) throws IOException{
 30 
 31         Configuration conf = new Configuration();
 32 
 33         FileSystem fs = FileSystem.get(conf);
 34 
 35         Path dstPath = new Path(dst); //目标路径
 36 
 37         //打开一个输出流
 38 
 39         FSDataOutputStream outputStream = fs.create(dstPath);
 40 
 41         outputStream.write(contents);
 42 
 43         outputStream.close();
 44 
 45         fs.close();
 46 
 47         System.out.println("文件创建成功！");
 48 
 49     }
 50 
 51     //上传本地文件
 52 
 53     public static void uploadFile(String src,String dst) throws IOException{
 54 
 55         Configuration conf = new Configuration();
 56 
 57         FileSystem fs = FileSystem.get(conf);
 58 
 59         Path srcPath = new Path(src); //原路径
 60 
 61         Path dstPath = new Path(dst); //目标路径
 62 
 63         //调用文件系统的文件复制函数,前面参数是指是否删除原文件，true为删除，默认为false
 64 
 65         fs.copyFromLocalFile(false,srcPath, dstPath);
 66 
 67         //打印文件路径
 68 
 69         System.out.println("Upload to "+conf.get("fs.default.name"));
 70 
 71         System.out.println("------------list files------------"+"
");
 72 
 73         FileStatus [] fileStatus = fs.listStatus(dstPath);
 74 
 75         for (FileStatus file : fileStatus)
 76 
 77         {
 78 
 79             System.out.println(file.getPath());
 80 
 81         }
 82 
 83         fs.close();
 84 
 85     }
 86 
 87     //文件重命名
 88 
 89     public static void rename(String oldName,String newName) throws IOException{
 90 
 91         Configuration conf = new Configuration();
 92 
 93         FileSystem fs = FileSystem.get(conf);
 94 
 95         Path oldPath = new Path(oldName);
 96 
 97         Path newPath = new Path(newName);
 98 
 99         boolean isok = fs.rename(oldPath, newPath);
100 
101         if(isok){
102 
103             System.out.println("rename ok!");
104 
105         }else{
106 
107             System.out.println("rename failure");
108 
109         }
110 
111         fs.close();
112 
113     }
114 
115     //删除文件
116 
117     public static void delete(String filePath) throws IOException{
118 
119         Configuration conf = new Configuration();
120 
121         FileSystem fs = FileSystem.get(conf);
122 
123         Path path = new Path(filePath);
124 
125         boolean isok = fs.deleteOnExit(path);
126 
127         if(isok){
128 
129             System.out.println("delete ok!");
130 
131         }else{
132 
133             System.out.println("delete failure");
134 
135         }
136 
137         fs.close();
138 
139     }
140 
141     //创建目录
142 
143     public static void mkdir(String path) throws IOException{
144 
145         Configuration conf = new Configuration();
146 
147         FileSystem fs = FileSystem.get(conf);
148 
149         Path srcPath = new Path(path);
150 
151         boolean isok = fs.mkdirs(srcPath);
152 
153         if(isok){
154 
155             System.out.println("create dir ok!");
156 
157         }else{
158 
159             System.out.println("create dir failure");
160 
161         }
162 
163         fs.close();
164 
165     }
166 
167     //读取文件的内容
168 
169     public static void readFile(String filePath) throws IOException{
170 
171         Configuration conf = new Configuration();
172 
173         FileSystem fs = FileSystem.get(conf);
174 
175         Path srcPath = new Path(filePath);
176 
177         InputStream in = null;
178 
179         try {
180 
181             in = fs.open(srcPath);
182 
183             IOUtils.copyBytes(in, System.out, 4096, false); //复制到标准输出流
184 
185         } finally {
186 
187             IOUtils.closeStream(in);
188 
189         }
190 
191     }
192 
193     public static void main(String[] args) throws IOException {
194 
195         //测试上传文件
196 
197         //uploadFile("/home/hadoop/music1.txt","hdfs://master:9000/user/hadoop/test.txt");
198 
199         //测试创建文件
200 
201       //byte[] contents =  "hello world 世界你好
".getBytes();
202 
203        //createFile("hdfs://master:9000/user/hadoop/test1/d.txt",contents);
204 
205         //测试重命名
206 
207         //rename("hdfs://master:9000/user/hadoop/test1/d.txt", "hdfs://master:9000/user/hadoop/test1/dd.txt");
208 
209         //测试删除文件
210 
211         //delete("hdfs://master:9000/user/hadoop/test1/d.txt"); //使用相对路径
212 
213        //delete("hdfs://master:9000/user/hadoop/test1");    //删除目录
214 
215         //测试新建目录
216 
217        // mkdir("hdfs://master:9000/user/hadoop/test1");
218 
219         //测试读取文件
220 
221         //readFile("hdfs://master:9000/user/hadoop/test1/d.txt");
222 
223     }
224 
225 }

View Code