storm源码编译及本地调试方法

发表于 2016-07-13 | 分类于 Storm | | 阅读次数

基础环境

IDE开发环境：intelliJIdea
JDK1.7 64bit
intelliJIdea安装maven插件，配置好仓库源
intelliJIdea安装clojure插件Cursive（需要注册并获取一个license，否则只能使用30天）
如果需要自己创建clojure项目进行开发，需要安装leiningen，下载地址

源码获取

从github checkout代码到本地即可，https://github.com/apache/storm.git

我这里编译的是我们目前正在用的0.10.0版本的代码。

导入idea及编译

打开idea，新建project，从源码导入，如下：

导入后，idea会自动根据pom.xml下载相关依赖包，部分依赖包如果下载不到，需要手动添加。完成后，可以看到project的module如下图所示：

阅读全文 »

kafka-0.10.0启动过程分析

发表于 2016-07-08 | 分类于 Kafka | | 阅读次数

kafka-0.10.0是官方出的最新稳定版本，提供了大量新的feature，具体可见这里，本文主要分析kafka-0.10-0的源码结构和启动过程。

源码结构

kafka-0.10.0的源码可以从github上fork一份，在源码目录下执行./gradlew idea生成idea项目，然后导入idea即可。这中间需要使用gradle进行依赖包的下载，导入后可以看到其源码结构如下图所示：

包括几大重要模块：

clients主要是kafka-client相关的代码，包括consumer、producer，还包括一些公共逻辑，如授权认证、序列化等。
connect主要是kafka-connect模块的代码逻辑，Kafka connect是0.9版本增加的特性,支持创建和管理数据流管道。通过它可以将大数据从其它系统导入到Kafka中，也可以从Kafka中导出到其它系统，比如数据库、elastic search等。
core模块是kafka的核心部分，主要包括broker的实现逻辑、producer和consumer的javaapi等。
streams模块主要是kafka-streaming的实现，提供了一整套描述常见流操作的高级语言API（比如 joining, filtering以及aggregation等），我们可以基于此开发流处理应用程序。

阅读全文 »

Storm的消息可靠处理机制

发表于 2015-11-05 | 分类于 Storm | | 阅读次数

简介

提交进入Storm运行的topology实际上是一个有向无环图（DAG），其中的节点是由spout和bolt组成，边则可以理解为消息从一个节点到传输到另一个节点的过程。对于spout产生的tuple，只有在topology上处理完毕后，才认为这个tuple被storm可靠处理。

Storm提供了可靠处理消息（storm中的通用名叫tuple）的框架，我们在写一个topology程序时，若需要保证spout产生的消息的可靠处理，需要做到两点：

第一是spout/bolt每生成一个新的tuple都告诉storm一下（其中spout发出的tuple有个id叫rootId），从而让storm能够追踪rootId和每个衍生tuple的处理状态。

第二是每个tuple被下游bolt处理完毕后，无论处理成功或失败，也再告诉storm一下，从而让storm知道是否需要spout重新发送rootId。

做了这两件事，storm就能知道这个tuple是否被处理完毕。如果是处理成功了的，就说明最初从spout发出的tuple（rootId）已在topology中处理完毕，无需spout重新发送。如果是处理失败的，storm则会告知spout重新发送rootId这个tuple。

在程序中实现消息可靠处理

那在写一个topology时，我们该如何做上面提到的两件事呢？

阅读全文 »

Spark on yarn的内存分配问题

发表于 2015-08-11 | 分类于 Spark | | 阅读次数

问题描述

在测试spark on yarn时，发现一些内存分配上的问题，具体如下。

在$SPARK_HOME/conf/spark-env.sh中配置如下参数：

SPARK_EXECUTOR_INSTANCES=4 在yarn集群中启动的executor进程数

SPARK_EXECUTOR_MEMORY=2G 为每个executor进程分配的内存大小

SPARK_DRIVER_MEMORY=1G 为spark-driver进程分配的内存大小

执行$SPARK_HOME/bin/spark-sql –master yarn，按yarn-client模式启动spark-sql交互命令行（即driver程序运行在本地，而非yarn的container中），日志显示的关于AppMaster和Executor的内存信息如下：

日志显示，AppMaster的内存是896MB，其中包含了384MB的memoryOverhead；启动了5个executor，第一个的可用内存是530.3MB，其余每个Executor的可用内存是1060.3MB。

到yarnUI看下资源使用情况，共启动了5个container，占用内存13G，其中一台NodeManager启动了2个container，占用内存4G（1个AppMaster占1G、另一个占3G），另外3台各启了1个container，每个占用3G内存。

再到sparkUI看下executors的情况，这里有5个executor，其中driver是运行在执行spark-sql命令的本地服务器上，另外4个是运行在yarn集群中。Driver的可用storage memory为530.3MB，另外4个都是1060.3MB（与日志信息一致）。

那么问题来了：

Yarn为container分配的最小内存由yarn.scheduler.minimum-allocation-mb参数决定，默认是1G，从yarnUI中看确实如此，可为何spark的日志里显示AppMaster的实际内存是896-384=512MB呢？384MB是怎么算出来的？
spark配置文件里指定了每个executor的内存为2G，为何日志和sparkUI上显示的是1060.3MB？
driver的内存配置为1G，为何sparkUI里显示的是530.3MB呢？
为何yarn中每个container分配的内存是3G，而不是executor需要的2G呢？

问题解析

进过一番调研，发现这里有些概念容易混淆，整理如下，序号对应上面的问题：

阅读全文 »

storm集群supervisor节点异常退出问题排查

发表于 2015-07-03 | 分类于 Storm | | 阅读次数

问题出现

测试storm集群为0.9.4版本，前段时间出现supervisor进程挂掉，而其上work进程仍然运行的诡异情况，通过日志看到supervisor进程挂掉之前出现以下异常：

问题排查过程

很明显，是commons-io包的FileUtils工具类抛出的异常，原因是在调用commons-io包的FileUtils工具类做move directory操作时，目的文件夹已存在。

查看调用代码（supervisor.clj的第374行），是调用download-storm-code方法从nimbus下载topology的代码，并且download-storm-code方法中做代码下载前加了锁避免并发写文件。

果然，这里没有判断stormroot文件夹是否已存在，是个bug，具体可见这个issue：https://issues.apache.org/jira/browse/STORM-805。

这个问题在0.9.5版本中随着STORM-130一起修复了，代码如下：

但这里有三个问题：

在调用download-storm-code方法前，代码中已做判断是否已下载topology代码，若已下载就不会调用download-storm-code方法了。为何进入这个方法后，做move directory操作时，代码却已经下载好了呢？
storm的历史发布版本有很多，为何0.9.4版本里会出现这个不该出现的问题，0.9.4相对老的版本是不是做了什么修改？
为何抛出异常后，supervisor进程就这么直接退出了？太弱了吧。。

阅读全文 »

交换空间使用率过高问题分析

发表于 2015-06-22 | 分类于问题分析 | | 阅读次数

问题现象

线上两台java后台服务每次上线后再过段时间，就出现swap空间使用率较高的现象，而jvm内存使用和垃圾回收情况则很正常。相关图表如下：

图中，每次上线后过一段时间，swap空间使用量会出现一个陡增，并随时间推移逐渐增加，期间会出现小幅度下降。

首先，从操作系统层面分析，swap空间使用较高，说明是系统物理内存不够用从而发生内存页交换，将部分内存数据搬至虚拟内存空间，也就是swap空间。但究竟是什么原因引起物理内存不足呢？因为Jvm堆大小是固定的，所以推断是因堆外内存占用空间较大引起。

于是，使用jmap -histo:live 把进程中的对象信息dump出来，dump信息如下：

确实发现存在大量DirectByteBuffer对象，这说明内存中确实有大量引用了堆外内存的对象没有被回收！

同时，内存中也对应存在着大量的sun.misc.Cleaner和java.nio.DirectByteBuffer$Deallocator对象。这两个类是用于回收堆外内存的。Cleaner对象是在DirectByteBuffer的构造函数中创建，其中封装了回收堆外内存的逻辑，Cleaner执行clean资源的操作是通过启动Deallocator线程实现的，这个线程把DirectByteBuffer对象引用的堆外内存做回收。

那么问题来了：

为什么DirectByteBuffer对象没有被回收？
怎么做才能让DirectByteBuffer对象能被及时回收？

问题分析

阅读全文 »

hadoop-yarn中ResourceManager的服务模块

发表于 2015-06-06 | 分类于 Hadoop | | 阅读次数

Yarn简述

Hadoop2.0引入了yarn（Yet Another Resource Negotiator）资源管理框架。1.0中的MapReduce计算框架变为运行在yarn上的一种application。

Yarn依然采用了master/slave结构，master是ResourceManager，负责整个集群的资源管理和调度，并且支持HA，slave是NodeManager，负责管理各子节点上的资源和任务。每个MapReduce作业提交给ResourceManager并被接受后，ResourceManager会通知某个NodeManager启动一个ApplicationMaster管理此作业的生命周期。

ResourceManager中的模块划分

Yarn中的大多数服务都是带状态的service实现，并通过事件驱动机制实现服务的状态转换和服务之间的交互。ResourceManager是yarn的核心组件，与NodeManager、ApplicationMaster、Client都有交互，提供了非常多的功能，下面基于hadoop2.7版本的实现，梳理一下ResourceManager中的重要service组件及其功能。

ResourceManager中按功能划分的service模块如下图所示。

ResourceManager中核心模块主要包括客户端交互模块、NodeManager管理模块、ApplicationMaster管理模块、Application管理模块、安全管理模块、以及资源管理模块（调度、预留）等。

各模块中的服务介绍

客户端交互模块：

AdminService
- 管理员可通过此接口管理集群，如更新节点、更新ACL、更新队列等。内部有个EmbeddedElectorService，如果RM启用了自动HA，则通过这个service做leader election。
ClientRMService
- 负责为客户端提供服务，是ApplicationClientProtocol协议的服务端。负责处理来自客户端的RPC请求，包括提交app、查询app运行状态、终止app等。
Webapp
- 提供web页面服务，展示集群状态和资源使用情况。

NodeManager管理模块

NMLivelinessMonitor
- 用于监控NM是否存活，若NM在一定时间内（默认10分钟）未上报心跳，则认为其挂了。
NodesListManager
- 负责维护节点列表，并动态加载白名单（yarn.resourcemanager.nodes.include-path）和黑名单（yarn.resourcemanager.nodes.exlude-path）节点。
RMNodeLabelsManager
- 负责节点的标签管理。
ResourceTrackerService
- 负责与NodeManager通信，处理来自NodeManager的请求，包括注册NodeManager和节点心跳两种。接口定义在ResourceTracker中。

ApplicationMaster管理模块

AMLivelinessMonitor：两个实例
- 用于监控ApplicationMaster是否正常，如果在指定时间内（默认10分钟）未收到AM的心跳，则认为其死掉了。
ApplicationMasterLauncher
- 负责通知某个NodeManager启动或销毁ApplicationMaster。在app请求被接受后，与某个NodeManager通信，告知其为此app启动相应的ApplicationMaster。若app运行结束或被kill，则通知app所在NodeManager销毁ApplicationMaster。其内部也维护了一个阻塞队列，并有一个后台线程异步处理提交进来的启动ApplicationMaster的请求。
ApplicationMasterService
- 负责与ApplicationMaster通信，是ApplicationMasterProtocol协议的服务端，ApplicationMaster在NodeManager上启动后通过此协议向ResourceManager注册自己，运行过程中向ResourceManager发送心跳，以及app运行结束后告知RM自己所在的container可以被释放了。

Application管理模块

RMAppManager
- ResourceManager接受客户端提交的app后，会通过RMAppManager来触发启动app的事件RMAppEventType.START，具体启动app的工作由RMAppImpl实现。
ApplicationACLsManager
- 负责app权限控制，包括查看和修改权限。
ContainerAllocationExpirer
- 用于监听NodeManager上是否正常启动了分配给ApplicationMaster的container，若在指定时间未启动（默认10分钟），ResourceManager会强制回收该container。
RMApplicationHistoryWriter
- 负责异步持久化Application运行中的相关日志，主要是Container、Application、ApplicationAttempt在启动和结束时的日志信息。

安全管理模块

RMSecretManagerService
- 负责管理各种通信密钥，包括：
  - RM与NM通信的NMTokenSecretManagerInRM
  - RM与container通信的RMContainerTokenSecretManager
  - 客户端与AM通信的ClientToAMTokenSecretManagerInRM
  - AM与RM通信的AMRMTokenSecretManager
  - DelegationTokenRenewer
  - 启用了安全时，负责定时更新认证token。

资源管理模块

ResourceScheduler
- 资源调度器，可通过yarn.resourcemanager.scheduler.class指定，ResourceManager默认使用的是org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler。
SchedulerEventDispatcher
- 用于处理SchedulerEventType类型的事件，其内部维护了一个存储SchedulerEvent的阻塞队列，并由一个后台线程从队列中取出资源请求事件，再调用ResourceScheduler进行处理。
ReservationSystem
- 资源预留系统，对应的实现有CapacityReservationSystem和FairReservationSystem。

此外，SystemMetricsPublisher负责发布RM的系统统计信息。AsyncDispatcher是中央事件处理分发器，ResourceManager启动时，通过它绑定了几种类型的事件的处理器，包括SchedulerEventType、RMAppEventType、ApplicationAttempt、RMAppAttemptEventType、RMNodeEventType、RMAppManagerEventType、AMLaunchEventType等。

上述各service在ResourceManager中的启动顺序为：

AsyncDispatcher
AdminService
RMActiveServices：是个CompositeService（即service列表，ResourceManager本身就是一个CompositeService），用于管理ResourceManager中的“活动”服务（必须在active的ResourceManager上启动的服务，启用HA时，备份ResourceManager上不启动这些服务），包括以下（按启动顺序）：
- RMSecretManagerService
- ContainerAllocationExpirer
- AMLivelinessMonitor
- RMNodeLabelsManager
- RMApplicationHistoryWriter
- SystemMetricsPublisher
- NodesListManager
- ResourceScheduler
- SchedulerEventDispatcher
- NMLivelinessMonitor
- ResourceTrackerService
- ApplicationMasterService
- ClientRMService
- ApplicationMasterLauncher
- DelegationTokenRenewer

mac系统下hadoop-2.7源码编译、导入eclipse及打包

发表于 2015-05-18 | 分类于 Hadoop | | 阅读次数

编译环境要求

JDK1.7+
MAVEN 3.0以上版本
如果需要编译native code，还需要CMake 2.6、Zlib devel、openssl devel（mac下一般安装了xcode后应该都会有这些包）。

编译方法

解压源码包hadoop-2.7.0-src.tar.gz，iterm下进入文件夹hadoop-2.7.0-src，然后根据需要执行相应的mvn命令就可以了。

仅编译：mvn compile
打包生成jar：mvn package
生成eclipse项目：eclipse:eclipse -DskipTests，加上-DskipTests可跳过test阶段。

期间遇到几个问题，记录如下。

问题记录

先执行mvn eclipse:eclipse -DskipTest生成eclipse项目，执行到一半时，提示下面的报错：

‘protoc –version’ did not return a version -> [Help 1]

意思也就是是找不到protoc命令，安装protocolbuffer后重试，又提示错误：

protoc version is ‘libprotoc 2.6.1′, expected version is ’2.5.0′

看上去是protocolbuffer版本问题，hadoop需要的版本是2.5.0，而系统安装的是2.6.1，查了很多资料，都是说protocolbuffer版本太低后来升到2.5的，而我这是2.6.1的版本，难不成还得降回去，不至于吧。因此，猜测这个版本限制是在pom.xml中写死的，于是grep了一下，发现果然在hadoop-project/pom.xml中配置了编译时使用的pb版本。

\2.5.0\

把以上配置项改为2.6.1，再重新执行生成eclipse项目的命令就OK了。

导入eclipse及打包

生成eclipse项目后，从eclipse里import existing project into workspace，选择hadoop-2.7.0-src目录，就会把所有代码模块导入eclipse了。接下来就可以看代码并修改了，比如增加一些日志信息等。

代码修改完毕后，可以再打出一个新的hadoop-distribution包来验证代码修改效果。

在hadoop-2.7.0-src目录下执行命令：mvn package -Pdist -Ptar -Pdocs -skipTests

等上漫长的一段时间，编译成功后，可以到hadoop-dist/target下找到新的jar包。

补充说明

编译过程需要从maven中央仓库下载大量依赖包，我使用的是oschina的库。

<mirror>
  <id>CN</id>
  <name>OSChina Central</name>
  <url>http://maven.oschina.net/content/groups/public/</url>
  <mirrorOf>central</mirrorOf>
</mirror>
  <profile>
	<id>oschina</id>
	<repositories>
		<repository>
			<id>nexus</id>
			<name>local private nexus</name>
			<url>http://maven.oschina.net/content/groups/public/</url>
			<releases>
				<enabled>true</enabled>
			</releases>
			<snapshots>
				<enabled>false</enabled>
			</snapshots>
		</repository>
	</repositories>
	<pluginRepositories>
		<pluginRepository>
			<id>nexus</id>
			<name>local private nexus</name>
			<url>http://maven.oschina.net/content/groups/public/</url>
			<releases>
				<enabled>true</enabled>
			</releases>
			<snapshots>
				<enabled>false</enabled>
			</snapshots>
		</pluginRepository>
	</pluginRepositories>
  </profile>

使用hexo+gitpage搭建博客

发表于 2014-09-02 | 分类于工具 | | 阅读次数

环境准备

系统：mac osx
软件：Node.js，npm，git，hexo
具体安装以及git与github打通的配置就不详述了，可以google到各种方法。

hexo命令

hexo init <folder> #表示执行init命令初始化hexo到你指定的目录

以下命令需要在<folder>目录下执行：
hexo generate #自动根据当前目录下文件,生成静态网页
hexo server #运行本地服务

启动服务后，就可以通过访问 http://localhost:4000 来看看效果了。
接下来，可以使用以下命令来创建一篇新博文：
hexo new “test blog 1”
创建一个名为test blog 1的博客页面，对应的md文件路径是<folder>/source/_posts\test blog 1.md

接下来就可以在这个md文件中写文章了，我使用的是MacDown来编辑md文件，支持实时查看页面效果，还是挺好用的。

发布博客

文章写好后，通过以下方式发布到github上。
1.编辑./_config.yml文件，修改以下部分，配置本地内容同步至github：

deploy:
  type: git
  repository: git@github.com:maohong/maohong.github.io.git
  branch: master

2.执行hexo generate(hexo g)生成html内容
3.执行hexo deploy(hexo d)讲更新内容发布至guthub

然后就可以访问主页查看效果了，可以使用github帐户名.github.io进行访问, 也可以设置个性域名。

redis服务端连接断开问题诊断

发表于 2014-06-01 | 分类于 Redis | | 阅读次数

问题现象

前段时间，由于线上redis服务器的内存使用率达到了机器总内存的50%以上，导致内存数据的dump持久化一直失败。扩展到多台redis后，应用系统访问redis时，在业务量较少时，时不时会出现以下异常，当业务量较大，redis访问频率很高时，却不会发生这个异常，一时觉得很诡异。

redis.clients.jedis.exceptions.JedisConnectionException: It seems like server has closed the connection.
at redis.clients.util.RedisInputStream.readLine(RedisInputStream.java:90) ~[jedis-2.1.0.jar:na]
at redis.clients.jedis.Protocol.processInteger(Protocol.java:110) ~[jedis-2.1.0.jar:na]
at redis.clients.jedis.Protocol.process(Protocol.java:70) ~[jedis-2.1.0.jar:na]
at redis.clients.jedis.Protocol.read(Protocol.java:131) ~[jedis-2.1.0.jar:na]
at redis.clients.jedis.Connection.getIntegerReply(Connection.java:188) ~[jedis-2.1.0.jar:na]
at redis.clients.jedis.Jedis.sismember(Jedis.java:1266) ~[jedis-2.1.0.jar:na]

看提示，应该是服务端主动关闭了连接。查看了新上线的redis服务器的配置，有这么一项：

# Close the connection after a client is idle for N seconds (0 to disable)
timeout 120

这项配置指的是客户端连接空闲超过多少秒后，服务端主动关闭连接，默认值0表示服务端永远不主动关闭。而op人员把服务器端的超时时间设置为了120秒。

这就解释了发生这个异常的原因。客户端使用了一个连接池管理访问redis的所有连接，这些连接是长连接，当业务量较小时，客户端部分连接使用率较低，当两次使用之间的间隔超过120秒时，redis服务端就主动关闭了这个连接，而等客户端下次再使用这个连接对象时，发现服务端已经关闭了连接，进而报错。

于是，再查看访问redis的系统（客户端）的配置：

客户端使用的是jedis内置的连接池，看其源码本质上是基于apache commons-pool实现的，其中有一个eviction线程，用于回收idle对象，对于redis连接池来说，也就是回收空闲连接。

JedisPoolConfig类继承自GenericObjectPoolConfig并覆盖了几项关于eviction线程的配置，具体如下：

_timeBetweenEvictionRunsMillis：eviction线程的运行周期。默认是-1，表示不启动eviction线程。这里设置为30秒。

_minEvictableIdleTimeMillis：对象处于idle状态的最长时间，默认是30分钟，这里设置为60秒。

通过客户端的默认配置看，对象的最大空闲时长是小于服务端的配置的，应该不是配置上的问题了。

于是，继续看是不是客户端代码使用上的问题。追踪到客户端代码如下：

可见，客户端首先尝试从本线程的ThreadLocal对象中获取jedis对象，若获取不到，再从masterJedisPool中取得jedis对象并放入ThreadLocal对象以便下次使用，并且jedis对象使用完毕后，没有从ThreadLocal中清除，也没有returnResource给masterJedisPool。

因此，问题产生的原因就在于此。ThreadLocal中的这个jedis对象被取出后没有return，对于对象池来说是处于非idle状态，因此不会被对象池evict。当业务量大时，这个jedis会被频繁使用，服务端认为这个jedis对应的连接是非空闲的，或者空闲时间达不到120秒，不会主动关闭，所以没什么问题。然而当业务量小时，这个jedis使用频率很低，当两次之间的使用间隔超出120秒时，服务端会主动把这个jedis的连接关闭，第二次调用时，就会出现上面的报错。

从代码开发者的角度来说，这么做的目的是避免频繁从pool中获取jedis对象和return jedis对象以提高性能。

解决方案有两个：

在redis-cli下在线修改redis 的配置，把timeout改回为0，无需重启redis即可直接生效，但redis若重启，配置会恢复。
修改客户端代码，使用完jedis对象后，从ThreadLocal中清除，再返回给连接池。

出于改动成本考虑，先采用了第一种方案，在线修改redis配置后，报错不再出现。