Hadoop集群上存储数据，同时基于MapReduce计算框架可以实现计算任务，那么无论是从数据保护的角度，还是从提交计算任务占用资源的角度来看，都需要存在一种权限管理与分配机制，能够很好地限制哪些人可以在HDFS上存储数据，哪些人可以利用集群的资源来处理特定的计算任务。当然，如果能够非常完美地解决这些问题是最好的。当前Hadoop本身提供的权限管理功能还不能满足普遍的需要，或者我们从Hadoop已有的一些简单或复杂的认证机制选择适合自己所在组织机构需要的，或者我们在外围开发一些权限管理系统与Hadoop整合作为补充。 对比Kerberos认证（Authentication）配置方式与SLA授权（Service Level Authorization）方式，Kerberos配置相当复杂，而且还要依赖于外部的密钥分发中心KDC（Key Distribution Center）服务器，如果KDC出现问题，那么就会导致依赖于KDC认证的整个Hadoop集群无法使用，鉴于此，对于一些相对小的开发团队来说还是更倾向于粗粒度的Hadoop SLA授权机制。 Hadoop SLA基于Hadoop的各种服务（基于协议来划分）与Linux系统的用户、用户组来实现。Had
按标签浏览文章: Hadoop
HDFS读文件过程分析：读取文件的Block数据
我们可以从java.io.InputStream类中看到，抽象出一个read方法，用来读取已经打开的InputStream实例中的字节，每次调用read方法，会读取一个字节数据，该方法抽象定义，如下所示： public abstract int read() throws IOException; Hadoop的DFSClient.DFSInputStream类实现了该抽象逻辑，如果我们清楚了如何从HDFS中读取一个文件的一个block的一个字节的原理，更加抽象的顶层只需要迭代即可获取到该文件的全部数据。 从HDFS读文件过程分析：获取文件对应的Block列表（http://shiyanjun.cn/archives/925.html）中，我们已经获取到一个文件对应的Block列表信息，打开一个文件，接下来就要读取实际的物理块数据，我们从下面的几个方面来详细说明读取数据的过程。 Client从Datanode读取文件的一个字节 下面，我们通过分析DFSClient.DFSInputStream中实现的代码，读取HDFS上文件的内容。首先从下面的方法开始： @Override public synchronized int read() throws IOException { int ret = read( oneByteBuf, 0, 1 ); return ( ret <= 0 ) ? -
HDFS写文件过程分析
HDFS是一个分布式文件系统，在HDFS上写文件的过程与我们平时使用的单机文件系统非常不同，从宏观上来看，在HDFS文件系统上创建并写一个文件，流程如下图（来自《Hadoop：The Definitive Guide》一书）所示： 具体过程描述如下： Client调用DistributedFileSystem对象的create方法，创建一个文件输出流（FSDataOutputStream）对象 通过DistributedFileSystem对象与Hadoop集群的NameNode进行一次RPC远程调用，在HDFS的Namespace中创建一个文件条目（Entry），该条目没有任何的Block 通过FSDataOutputStream对象，向DataNode写入数据，数据首先被写入FSDataOutputStream对象内部的Buffer中，然后数据被分割成一个个Packet数据包 以Packet最小单位，基于Socket连接发送到按特定算法选择的HDFS集群中一组DataNode（正常是3个，可能大于等于1）中的一个节点上，在这组DataNode组成的Pipeline上依次传输Packet 这组DataNode组成的Pipeline反方向上，发送ack，最终由Pipeline中第一个DataNode节点将Pipeline ack发送给Client 完成向文件写入数据，Client在文件输出流（FSD
HDFS读文件过程分析：获取文件对应的Block列表
在使用Java读取一个文件系统中的一个文件时，我们会首先构造一个DataInputStream对象，然后就能够从文件中读取数据。对于存储在HDFS上的文件，也对应着类似的工具类，但是底层的实现逻辑却是非常不同的。我们先从使用DFSClient.DFSDataInputStream类来读取HDFS上一个文件的一段代码来看，如下所示： package org.shirdrn.hadoop.hdfs; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; public class HdfsFileReader { public static void main(String[] args) { String file = "hdfs://hadoop-cluster-m:8020/data/logs/basis_user_behavior/201405071237_10_10_1_73.log"; Path path = new Path(file); Configuration conf = new Configuration();
HDFS格式化过程分析
我们知道，Namenode启动时可以指定不同的选项，当指定-format选项时，就是格式化Namenode，可以在Namenode类中看到格式化的方法，方法签名如下所示： private static boolean format(Configuration conf, boolean isConfirmationNeeded, boolean isInteractive) throws IOException 在该方法中，首先调用FSNamesystem类的方法，获取到待格式化的name目录和edit目录： Collection<File> editDirsToFormat = Collection<File> dirsToFormat = FSNamesystem.getNamespaceDirs(conf); FSNamesystem.getNamespaceEditsDirs(conf); 跟踪调用FSNamesystem类的方法，可以看到，实际上获取到的目录为： name目录：是根据配置的dfs.name.dir属性，如果没有配置，默认使用目录/tmp/hadoop/dfs/name。 edit目录：是根据配置的dfs.name.edits.dir属性，如果没有配置，默认使用目录/tmp/hadoop/dfs/name。 在上面format方法中，创建对应的name目录和edit目录，对应如下代码行： FSNamesystem nsys = new FSNamesystem(new FSImage(dirsToFormat,
Sqoop-1.4.4安装配置及基本使用
Sqoop是Apache旗下的开源项目，能够在Hadoop和结构化存储系统之间进行数据传输（导入、导出），尤其是当下应用非常广泛的关系型数据库。通常，可能很多业务数据都存储在关系型数据库中，当数据规模达到一定程度后，使用关系型数据对数据进行分析可能会存在一定的瓶颈，如上亿级别记录的复杂统计分析等。将关系型数据库中的数据同步到Hadoop平台上，借助Hadoop平台的可扩展的优势，可以进行复杂的统计分析，处理效率会有一定程度的提升。 下面，我们通过安装配置Sqoop，来体验一下Sqoop在Hadoop和MySQL之间进行数据同步的特性。 准备和配置 我们在使用的主机及其应用进程部署情况，如下所示： 节点m1（10.95.3.56）：Sqoop-1.4.4，Hive-0.12.0，Namenode、JobTracker MySQL节点（10.95.3.49）：MySQL数据库 我们先验证Sqoop能够成功连接MySQL数据库，然后验证将MySQL数据库表中的数据，同步到Hive中。 首先，在CentOS 6.4下安装MySQL数据库（服务器IP为：10.95.3.49）： rpm -qa | grep mysql sudo rpm -e --nodeps mysql yum list | grep mysql sudo yum ins
Hive JOIN使用详解
Hive是基于Hadoop平台的，它提供了类似SQL一样的查询语言HQL。有了Hive，如果使用过SQL语言，并且不理解Hadoop MapReduce运行原理，也就无法通过编程来实现MR，但是你仍然可以很容易地编写出特定查询分析的HQL语句，通过使用类似SQL的语法，将HQL查询语句提交Hive系统执行查询分析，最终Hive会帮你转换成底层Hadoop能够理解的MR Job。 对于最基本的HQL查询我们不再累述，这里主要说明Hive中进行统计分析时使用到的JOIN操作。在说明Hive JOIN之前，我们先简单说明一下，Hadoop执行MR Job的基本过程（运行机制），能更好的帮助我们理解HQL转换到底层的MR Job后是如何执行的。我们重点说明MapReduce执行过程中，从Map端到Reduce端这个过程（Shuffle）的执行情况，如图所示（来自《Hadoop: The Definitive Guide》）： 基本执行过程，描述如下： 一个InputSplit输入到map，会运行我们实现的Mapper的处理逻辑，对数据进行映射操作。 map输出时，会首先将输出中间结果写入到map自带的buffer中（buffer默认大小为100M，可以通过io.sort.mb配置）。 map自带的buffer使用容
Hadoop-2.2.0集群安装配置实践
Hadoop 2.x和1.x已经大不相同了，应该说对于存储计算都更加通用了。Hadoop 2.x实现了用来管理集群资源的YARN框架，可以面向任何需要使用基于HDFS存储来计算的需要，当然MapReduce现在已经作为外围的插件式的计算框架，你可以根据需要开发或者选择合适的计算框架。目前，貌似对MapReduce支持还是比较好的，毕竟MapReduce框架已经还算成熟。其他一些基于YARN框架的标准也在开发中。 YARN框架的核心是资源的管理和分配调度，它比Hadoop 1.x中的资源分配的粒度更细了，也更加灵活了，它的前景应该不错。由于极大地灵活性，所以在使用过程中由于这些配置的灵活性，可能使用的难度也加大了一些。另外，我个人觉得，YARN毕竟还在发展之中，也有很多不成熟的地方，各种问题频频出现，资料也相对较少，官方文档有时更新也不是很及时，如果我选择做海量数据处理，可能YARN还不能满足生产环境的需要。如果完全使用MapReduce来做计算，还是选择相对更加成熟的Hadoop 1.x版本用于生产环境。 下面使用4台机器，操作系统为CentOS 6.4 64位，一台做主节点，另外三台做从节点，实践集
Hadoop Job使用第三方依赖jar文件
当我们实现了一个Hadoop MapReduce Job以后，而这个Job可能又依赖很多外部的jar文件，在Hadoop集群上运行时，有时会出现找不到具体Class的异常。出现这种问题，基本上就是在Hadoop Job执行过程中，没有从执行的上下文中找到对应的jar文件（实际是unjar的目录，目录里面是对应的Class文件）。所以，我们自然而然想到，正确配置好对应的classpath，MapReduce Job运行时就能够找到。 有两种方式可以更好地实现，一种是设置HADOOP_CLASSPATH，将Job所依赖的jar文件加载到HADOOP_CLASSPATH，这种配置只针对该Job生效，Job结束之后HADOOP_CLASSPATH会被清理；另一种方式是，直接在构建代码的时候，将依赖jar文件与Job代码打成一个jar文件，这种方式可能会使得最终的jar文件比较大，但是结合一些代码构建工具，如Maven，可以在依赖控制方面保持一个Job一个依赖的构建配置，便于管理。下面，我们分别说明这两种方式。 设置HADOOP_CLASSPATH 比如，我们有一个使用HBase的应用，操作HBase数据库中表，肯定需要ZooKeeper，所以对应的jar文件的位置都要设置正确，让运行时Job能够
Hadoop Streaming原理及实践
Hadoop Streaming提供了一个便于进行MapReduce编程的工具包，使用它可以基于一些可执行命令、脚本语言或其他编程语言来实现Mapper和 Reducer，从而充分利用Hadoop并行计算框架的优势和能力，来处理大数据。需要注意的是，Streaming方式是基于Unix系统的标准输入输出来进行MapReduce Job的运行，它区别与Pipes的地方主要是通信协议，Pipes使用的是Socket通信，是对使用C++语言来实现MapReduce Job并通过Socket通信来与Hadopp平台通信，完成Job的执行。任何支持标准输入输出特性的编程语言都可以使用Streaming方式来实现MapReduce Job，基本原理就是输入从Unix系统标准输入，输出使用Unix系统的标准输出。 Hadoop是使用Java语言编写的，所以最直接的方式的就是使用Java语言来实现Mapper和Reducer，然后配置MapReduce Job，提交到集群计算环境来完成计算。但是很多开发者可能对Java并不熟悉，而是对一些具有脚本特性的语言，如C++、Shell、Python、 Ruby、PHP、Perl有实际开发经验，Hadoop Streaming为这一类开发者提供了使用Hadoop集群来进行处理数据的工具，即工具包
Hadoop MapReduce处理海量小文件：压缩文件
在HDFS上存储文件，大量的小文件是非常消耗NameNode内存的，因为每个文件都会分配一个文件描述符，NameNode需要在启动的时候加载全部文件的描述信息，所以文件越多，对 NameNode来说开销越大。 我们可以考虑，将小文件压缩以后，再上传到HDFS中，这时只需要一个文件描述符信息，自然大大减轻了NameNode对内存使用的开销。MapReduce计算中，Hadoop内置提供了如下几 种压缩格式： DEFLATE gzip bzip2 LZO 使用压缩文件进行MapReduce计算，它的开销在于解压缩所消耗的时间，在特定的应用场景中这个也是应该考虑的问题。不过对于海量小文件的应用场景，我们压缩了小文件，却换 来的Locality特性。 假如成百上千的小文件压缩后只有一个Block，那么这个Block必然存在一个DataNode节点上，在计算的时候输入一个InputSplit，没有网络间传输数据的开销，而且是在本地进行 运算。倘若直接将小文件上传到HDFS上，成百上千的小Block分布在不同DataNode节点上，为了计算可能需要“移动数据”之后才能进行计算。文件很少的情况下，除了NameNode内 存使用开销以外，可能感觉不到网
Hadoop MapReduce处理海量小文件：基于CombineFileInputFormat
在使用Hadoop处理海量小文件的应用场景中，如果你选择使用CombineFileInputFormat，而且你是第一次使用，可能你会感到有点迷惑。虽然，从这个处理方案的思想上很容易理解，但是可能会遇到这样那样的问题。 使用CombineFileInputFormat作为Map任务的输入规格描述，首先需要实现一个自定义的RecordReader。 CombineFileInputFormat的大致原理是，他会将输入多个数据文件（小文件）的元数据全部包装到CombineFileSplit类里面。也就是说，因为小文件的情况下，在HDFS中都是单Block的文件，即一个文件一个Block，一个CombineFileSplit包含了一组文件Block，包括每个文件的起始偏移（offset），长度（length），Block位置（localtions）等元数据。如果想要处理一个CombineFileSplit，很容易想到，对其包含的每个InputSplit（实际上这里面没有这个，你需要读取一个小文件块的时候，需要构造一个FileInputSplit对象）。 在执行MapReduce任务的时候，需要读取文件的文本行（简单一点是文本行，也可能是其他格式数据）。那么对于CombineFileSplit来说，你需要处理其包含的小文