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方式访问
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具体操作
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
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 }