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,最终

Kafka+Storm+HDFS整合实践

在基于Hadoop平台的很多应用场景中,我们需要对数据进行离线和实时分析,离线分析可以很容易地借助于Hive来实现统计分析,但是对于实时的需求Hive就不合适了。实时应用场景可以使用Storm,它是一个实时处理系统,它为实时处理类应用提供了一个计算模型,可以很容易地进行编程处理。为了统一离线和实时计算,一般情况下,我们都希望将离线和实时计算的数据源的集合统一起来作为输入,然后将数据的流向分别经由实时系统和离线分析系统,分别进行分析处理,这时我们可以考虑将数据源(如使用Flume收集日志)直接连接一个消息中间件,如Kafka,可以整合Flume+Kafka,Flume作为消息的Producer,生产的消息数据(日志数据、业务请求数据等等)发布到Kafka中,然后通过订阅的方式,使用Storm的Topology作为消息的Consumer,在Storm集群中分别进行如下两个需求场景的处理: 直接使用Storm的Topology对数据进行实时分析处理 整合Storm+HDFS,将消息处理后写入HDFS进行

Flume(NG)架构设计要点及配置实践

Flume NG是一个分布式、可靠、可用的系统,它能够将不同数据源的海量日志数据进行高效收集、聚合、移动,最后存储到一个中心化数据存储系统中。由原来的Flume OG到现在的Flume NG,进行了架构重构,并且现在NG版本完全不兼容原来的OG版本。经过架构重构后,Flume NG更像是一个轻量的小工具,非常简单,容易适应各种方式日志收集,并支持failover和负载均衡。 架构设计要点 Flume的架构主要有一下几个核心概念: Event:一个数据单元,带有一个可选的消息头 Flow:Event从源点到达目的点的迁移的抽象 Client:操作位于源点处的Event,将其发送到Flume Agent Agent:一个独立的Flume进程,包含组件Source、Channel、Sink Source:用来消费传递到该组件的Event Channel:中转Event的一个临时存储,保存有Source组件传递过来的Event Sink:从Channel中读取并移除Event,将Event传递到Flow Pipeline中的下一个Agent(如果有的话) Flume NG架构,如图所示: 外

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";

Hive-0.12.0的Web接口HWI安装、配置、改造及使用

使用Hive的HWI接口,可以通过在Web页面上提交HQL查询操作,并浏览查询结果数据。默认情况下,HWI只支持浏览结果数据,不能够下载查询结果文件(当然,HWI可能也是考虑通过Web下载大量的结果数据,对服务器造成压力,或者处于安全方面的考虑)。我们对HWI进行了简单的改造,改造内容主要是增加了一个内置的文件服务器,可以通过页面进行查询,然后下载结果文件。 HWI安装配置 首先,要保证Hadoop集群正常运行,Hive能够正常使用。 先要安装Ant,如下所示: wget http://mirrors.hust.edu.cn/apache//ant/binaries/apache-ant-1.9.4-bin.tar.gz tar xvzf apache-ant-1.9.4-bin.tar.gz ln -s /usr/local/apache-ant-1.9.4-bin /usr/local/ant 修改Hive的环境配置文件hive-env.sh,增加如下配置内容: export ANT_LIB=/usr/local/ant 将如下JAR文件拷贝到${HIVE_HOME}/lib目录下面: // 用于编译JSP文件 jasper-compiler-5.5.23.jar jasper-runtim

Spring+Mybatis实现动态SQL查询

在报表类应用中,通常需要根据不同的维度去组合复杂的查询条件,然后构造SQL去执行查询。如果只是通过在程序中简单地拼接SQL语句,工作量会非常大,而且代码可能也非常难以维护。Mybatis支持动态SQL查询功能,可以通过配置动态的SQL来简化程序代码中复杂性,不过,这个颇有点XML编程的韵味,通过XML来处理复杂的数据判断、循环的功能,其实也很好理解。 准备工作 下面,我们首先创建一个MySQL示例表,如下所示: CREATE TABLE `traffic_info` ( `id` int(11) NOT NULL AUTO_INCREMENT, `domain` varchar(64) NOT NULL, `traffic_host` varchar(64) NOT NULL, `month` varchar(8) NOT NULL, `monthly_traffic` int(11) DEFAULT '0', `global_traffic_rank` int(11) DEFAULT '0', `native_traffic_rank` int(11) DEFAULT '0', `rank_in_country` varchar(64) DEFAULT NULL, `address` varchar(200) DEFAULT NULL, `email` var

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方法中,创建对应的na

Shark-0.9.0安装配置运行实践

Shark(Hive on Spark)是UC Lab为Spark设计并开源的一款数据仓库系统,提供了分布式SQL查询引擎,它能够完全兼容Hive。首先,我们通过下面的图,看一下Shark与Hive的关系(http://shark.cs.berkeley.edu/img/shark-hive-integration.png): 以前我们使用Hive分析HDFS中数据时,通过将HQL翻译成MapReduce作业(Job)在Hadoop集群上运行;而使用Shark可以像使用Hive一样容易,如HQL、Metastore、序列化格式、UDF等Shark都支持,不同的是Shark运行在Spark集群上执行计算,基于Spark系统所使用的RDD模型。官方文档给出的性能方面的数据是,使用Shark查询分析HDFS数据,能比Hive快30多倍,如图所示(http://shark.cs.berkeley.edu/img/perf.png): 下面,我们通过安装配置Shark来简单地体验一下。 准备软件包 jdk-7u25-linux-x64.tar.gz scala-2.10.3.tgz apache-maven-3.2.1-bin.tar.gz hadoop-1.2.1.tar.gz spark-0.9.0-incubating-bin-hadoop1.tgz hive-

RDD:基于内存的集群计算容错抽象

该论文来自Berkeley实验室,英文标题为:Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing。下面的翻译,我是基于科学网翻译基础上进行优化、修改、补充,这篇译文翻译得很不错。在此基础上,我增加了来自英文原文的图和表格数据,以及译文中缺少的未翻译的部分。如果翻译措辞或逻辑有误,欢迎批评指正。 摘要 本文提出了分布式内存抽象的概念——弹性分布式数据集(RDD,Resilient Distributed Datasets),它具备像MapReduce等数据流模型的容错特性,并且允许开发人员在大型集群上执行基于内存的计算。现有的数据流系统对两种应用的处理并不高效:一是迭代式算法,这在图应用和机器学习领域很常见;二是交互式数据挖掘工具。这两种情况下,将数据保存在内存中能够极大地提高性能。为了有效地实现容错,RDD提供了一种高度受限的共享内存,即RDD是只读的,并且只能通过其他RDD上的批量操作来创建。尽管如此,

使用Java编写并运行Spark应用程序

我们首先提出这样一个简单的需求: 现在要分析某网站的访问日志信息,统计来自不同IP的用户访问的次数,从而通过Geo信息来获得来访用户所在国家地区分布状况。这里我拿我网站的日志记录行示例,如下所示: 121.205.198.92 - - [21/Feb/2014:00:00:07 +0800] "GET /archives/417.html HTTP/1.1" 200 11465 "http://shiyanjun.cn/archives/417.html/" "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0" 121.205.198.92 - - [21/Feb/2014:00:00:11 +0800] "POST /wp-comments-post.php HTTP/1.1" 302 26 "http://shiyanjun.cn/archives/417.html/" "Mozilla/5.0 (Windows NT 5.1; rv:23.0) Gecko/20100101 Firefox/23.0" 121.205.198.92 - - [21/Feb/2014:00:00:12 +0800] "GET /archives/417.html/ HTTP/1.1" 301 26 "http://shiyanjun.cn/archives/417.

CentOS 6.4下安装配置Spark-0.9集群

Spark是一个快速、通用的计算集群框架,它的内核使用Scala语言编写,它提供了Scala、Java和Python编程语言high-level API,使用这些API能够非常容易地开发并行处理的应用程序。 下面,我们通过搭建Spark集群计算环境,并进行简单地验证,来体验一下使用Spark计算的特点。无论从安装运行环境还是从编写处理程序(用Scala,Spark默认提供的Shell环境可以直接输入Scala代码进行数据处理),我们都会觉得比Hadoop MapReduce计算框架要简单得多,而且,Spark可以很好地与HDFS进行交互(从HDFS读取数据,以及写数据到HDFS中)。 安装配置 下载安装配置Scala wget http://www.scala-lang.org/files/archive/scala-2.10.3.tgz tar xvzf scala-2.10.3.tgz 在~/.bashrc中增加环境变量SCALA_HOME,并使之生效: export SCALA_HOME=/usr/scala/scala-2.10.3 export PATH=$PATH:$SCALA_HOME/bin 下载安装配置Spark 我们首先在主节点m1上配置Spark程序,然后

Oozie Coordinator使用及详解

Oozie所支持工作流,工作流定义通过将多个Hadoop Job的定义按照一定的顺序组织起来,然后作为一个整体按照既定的路径运行。一个工作流已经定义了,通过启动该工作流Job,就会执行该工作流中包含的多个Hadoop Job,直到完成,这就是工作流Job的生命周期。 那么,现在我们有一个工作流Job,希望每天半夜00:00启动运行,我们能够想到的就是通过写一个定时脚本来调度程序运行。如果我们有多个工作流Job,使用crontab的方式调用可能需要编写大量的脚本,还要通过脚本来控制好各个工作流Job的执行时序问题,不但脚本不好维护,而且监控也不方便。基于这样的背景,Oozie提出了Coordinator的概念,他们能够将每个工作流Job作为一个动作(Action)来运行,相当于工作流定义中的一个执行节点(我们可以理解为工作流的工作流),这样就能够将多个工作流Job组织起来,称为Coordinator Job,并指定触发时间和频率,还可以配置数据集、并发数等。一个Coordinator Job包含了