Apache Storm内部原理分析

本文算是个人对Storm应用和学习的一个总结,由于不太懂Clojure语言,所以无法更多地从源码分析,但是参考了官网、好多朋友的文章,以及《Storm Applied: Strategies for real-time event processing》这本书,以及结合自己使用Storm的经历,希望对于想深入一点了解Storm原理的朋友能有所帮助,有不足之处欢迎拍砖交流。 Storm集群架构 Storm集群采用主从架构方式,主节点是Nimbus,从节点是Supervisor,有关调度相关的信息存储到ZooKeeper集群中,架构如下图所示: 具体描述,如下所示: Nimbus Storm集群的Master节点,负责分发用户代码,指派给具体的Supervisor节点上的Worker节点,去运行Topology对应的组件(Spout/Bolt)的Task。 Supervisor Storm集群的从节点,负责管理运行在Supervisor节点上的每一个Worker进程的启动和终止。通过Storm的配置文件中的supervisor.slots.ports配置项,可以指定在一个Supervisor上最大允许多少个Slot,每个Slot通过

MapReduce V1:MapTask执行流程分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。 在文章《MapReduce V1:TaskTracker设计要点概要分析》中我们已经了解了org.apache.hadoop.mapred.Child启动的基本流程,在Child VM启动的过程中会运行MapTask,实际是运行用户编写的MapReduce程序中的map方法中的处理逻辑,我们首先看一下,在Child类中,Child基于TaskUmbilicalProtocol协议与TaskTracker通信,获取到该Child VM需要加载的Task相关数据,包括Task本身,代码如下所示: final TaskUmbilicalProtocol umbilical = taskOwner.doAs(new PrivilegedExceptionAction<TaskUmbilicalProtocol>() { @Override public TaskUmbilicalProtocol run() throws Exception { // 建立Child到TaskTracker的RPC连接 return (TaskUmbilicalProtocol)RPC.getProxy(TaskUmbilicalProtocol.class, TaskUmbilicalProtocol.versionID,

基于协同过滤的推荐方法

协同过滤(Collaborative Filtering, CF)是推荐系统广泛使用的一种技术,它主要通过考虑用户(User)与用户之间、物品(Item)与物品之间的相似度(Similarity),来向用户推荐物品,常被用在电商网站中。其中,在推荐系统中最常使用的协同过滤方法,有如下4种: 基于用户的协同过滤推荐 基于物品的协同过滤推荐 基于模型的协同过滤推荐 混合协同过滤推荐 上面4种方法中,基于用户的协同过滤推荐、基于物品的协同过滤推荐都是基于内存的协同过滤推荐,一般在数据量较小的应用场景下,可以直接在线使用的实时推荐方法;基于模型的协同过滤推荐一般用于离线计算,它采用机器学习的方法,一般首相将用户偏好行为数据分成2个数据集(有时可能会将数据集分成k个子集,采用交叉验证的方式来提高模型精度),一个为训练集,一个为测试集,使用训练集数据来训练出推荐模型,然后使用测试集数据来评估模型的精度,当满足特定精度时,可以将得到的推荐模型应用于实

MapReduce V1:TaskTracker端启动Task流程分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。 TaskTracker周期性地向JobTracker发送心跳报告,在RPC调用返回结果后,解析结果得到JobTracker下发的运行Task的指令,即LaunchTaskAction,就会在TaskTracker节点上准备运行这个Task。Task的运行是在一个与TaskTracker进程隔离的JVM实例中执行,该JVM实例是通过org.apache.hadoop.mapred.Child来创建的,所以在创建Child VM实例之前,需要做大量的准备工作来启动Task运行。一个Task的启动过程,如下序列图所示: 通过上图,结合源码,我们将一个Task启动的过程,分为下面3个主要的步骤: 初始化跟踪Task运行的相关数据结构 准备Task运行所共享的Job资源 启动Task 下面,我们详细分析上面3个步骤的流程: 初始化跟踪Task运行的相关数据结构 如果是LaunchTaskAction,则TaskTracker会将该指令加入到一个启动Task的队列中,进行一步加载处理,如下所示: private void addToTaskQueue(LaunchTaskA

MapReduce V1:TaskTracker设计要点概要分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。 本文不打算深入地详细分析TaskTracker某个具体的处理流程,而是概要地分析TaskTracker在MapReduce框架中的主要负责处理那些事情,是我们能够在宏观上了解TaskTracker端都做了哪些工作。我尽量将TaskTracker端的全部要点内容提出来,但是涉及到详细的分析,只是点到为止,后续会对相应模块的处理流程结合代码进行分析。 TaskTracker主要负责MapReduce计算集群中Task运行的管理,所以TaskTracker要管理的事情比较多。一个MapReduce Job由很多的Task组成,而一个Job的所有Task被分成几个相斥的子集,每个子集被分配到某一个TaskTracker上去运行,所以一个TaskTracker管理运行了一个Job的所有Task的一个子集,也就是说TaskTracker不仅要维护每个Job对应的一个Task的子集,还要维护这些Task所属的Job的运行状态,对于Job/Task的状态的管理都是与JobTracker通过RPC通信保持状态的同步。 下面是TaskTracker端

k-medoids聚类算法实现

k-medoids聚类算法,即k-中心聚类算法,它是基于k-means聚类算法的改进。我们知道,k-means算法执行过程,首先需要随机选择初始质心,只有第一次随机选择的初始质心才是实际待聚类点集中的点,而后续将非质心点指派到对应的质心点后,重新计算得到的质心并非是待聚类点集中的点,而且如果某些非质心点是离群点的话,导致重新计算得到的质心可能偏离整个簇,为了解决这个问题,提出了改进的k-medoids聚类算法。 k-medoids聚类算法也是通过划分的方式来计算得到聚类结果,它使用绝对差值和(Sum of Absolute Differences,SAD)的度量来衡量聚类结果的优劣,在n维欧几里德空间中,计算SAD的公式如下所示: 围绕中心点划分(Partitioning Around Medoids,PAM)的方法是比较常用的,使用PAM方法进行处理,可以指定一个最大迭代次数的参数,在迭代过程中基于贪心策略来选择使得聚类的质量最高的划分。使用PAM的方法处理,每次交换一个中心点和非中心点,然后执行

Bisecting k-means聚类算法实现

Bisecting k-means聚类算法,即二分k均值算法,它是k-means聚类算法的一个变体,主要是为了改进k-means算法随机选择初始质心的随机性造成聚类结果不确定性的问题,而Bisecting k-means算法受随机选择初始质心的影响比较小。 首先,我们考虑在欧几里德空间中,衡量簇的质量通常使用如下度量:误差平方和(Sum of the Squared Error,简称SSE),也就是要计算执行聚类分析后,对每个点都要计算一个误差值,即非质心点到最近的质心的距离。那么,既然每个非质心点都已经属于某个簇,也就是要计算每个非质心点到其所在簇的质心的距离,最后将这些距离值相加求和,作为SSE去评估一个聚类的质量如何。我们的最终目标是,使得最终的SSE能够最小,也就是一个最小化目标SSE的问题。在n维欧几里德空间,SSE形式化地定义,计算公式如下: Bisecting k-means聚类算法的基本思想是,通过引入局部二分试验,每次试验都通过二分具有最大SSE值的一个簇,二分这个簇以后得到的

k-means聚类算法原理及其实现

k-means(k-均值)算法是一种基于距离的聚类算法,它用质心(Centroid)到属于该质心的点距离这个度量来实现聚类,通常可以用于N维空间中对象。下面,我们以二维空间为例,概要地总结一下k-means聚类算法的一些要点: 除了随机选择的初始质心,后续迭代质心是根据给定的待聚类的集合S中点计算均值得到的,所以质心一般不是S中的点,但是标识的是一簇点的中心。 基本k-means算法,开始需要随机选择指定的k个质心,因为初始k个质心是随机选择的,所以每次执行k-means聚类的结果可能都不相同。如果初始随机选择的质心位置不好,可能造成k-means聚类的结果非常不理想。 计算质心:假设k-means聚类过程中,得到某一个簇的集合Ci={p(x1,y1), p(x2,y2), …,p(xn,yn)},则簇Ci的质心,质心x坐标为(x1+x2+ …+xn)/n,质心y坐标为(y1+y2+ …+yn)/n。 k-means算法的终止条件:质心在每一轮迭代中会发生变化,然后需要重新将非质心点指派给最近的质心而形

ElasticSearch-2.0.0集群安装配置与API使用实践

ElasticSearch是基于全文搜索引擎库Lucene构建的分布式搜索引擎,我们可以直接使用ElasticSearch实现分布式搜索系统的搭建与使用,都知道,Lucene只是一个搜索框架,它提供了搜索引擎操作的基本API,如果要实现一个能够使用的搜索引擎系统,还需要自己基于Lucene的API去实现,工作量很大,而且还需要很好地掌握Lucene的底层实现原理。 ElasticSearch是一个完整的分布式搜索引擎系统,它的一些基本特性包括如下: 全文检索 提供插件机制,可以共享重用插件的功能 分布式文件存储 分布式实时索引和搜索 实时统计分析 可以横向扩展,支持大规模数据的搜索 简单易用的RESTful API 基于Replication实现了数据的高可用特性 与其他系统的集成 支持结构化和非结构化数据 灵活的Schema设计(Mappings) 支持多编程语言客户端 我个人感觉,ElasticSearch尽量屏蔽底层Lucene相关的技术细节,让你根本无从感觉底层Lucene相关的内容,这样你可以省去了了解Lucene 的成本

MapReduce V1:JobTracker处理Heartbeat流程分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。这篇文章的内容,更多地主要是描述处理/交互流程性的东西,大部分流程图都是经过我梳理后画出来的(开始我打算使用序列图来描述流程,但是发现很多流程在单个对象内部都已经非常复杂,想要通过序列图表达有点担心描述不清,所以选择最基本的程序流程图),可能看起来比较枯燥,重点还是关注主要的处理流程要点,特别的地方我会刻意标示出来,便于理解。 JobTracker与TaskTracker之间通过org.apache.hadoop.mapred.InterTrackerProtocol协议来进行通信,TaskTracker通过该接口进行远程调用实现Heartbeat消息的发送,协议方法定义如下所示: HeartbeatResponse heartbeat(TaskTrackerStatus status, boolean restarted, boolean initialContact, boolean acceptNewTasks,

DBSCAN聚类算法原理及其实现

DBSCAN(Density-Based Spatial Clustering of Applications with Noise)聚类算法,它是一种基于高密度连通区域的、基于密度的聚类算法,能够将具有足够高密度的区域划分为簇,并在具有噪声的数据中发现任意形状的簇。我们总结一下DBSCAN聚类算法原理的基本要点: DBSCAN算法需要选择一种距离度量,对于待聚类的数据集中,任意两个点之间的距离,反映了点之间的密度,说明了点与点是否能够聚到同一类中。由于DBSCAN算法对高维数据定义密度很困难,所以对于二维空间中的点,可以使用欧几里德距离来进行度量。 DBSCAN算法需要用户输入2个参数:一个参数是半径(Eps),表示以给定点P为中心的圆形邻域的范围;另一个参数是以点P为中心的邻域内最少点的数量(MinPts)。如果满足:以点P为中心、半径为Eps的邻域内的点的个数不少于MinPts,则称点P为核心点。 DBSCAN聚类使用到一个k-距离的概念,k-距离是指:给定数据集P={p(i); i=0,1,…n},对于任意点P(i

MapReduce V1:JobTracker端Job/Task数据结构

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。在MapReduce程序运行的过程中,JobTracker端会在内存中维护一些与Job/Task运行相关的信息,了解这些内容对分析MapReduce程序执行流程的源码会非常有帮助。 在编写MapReduce程序时,我们是以Job为单位进行编程处理,一个应用程序可能由一组Job组成,而MapReduce框架给我们暴露的只是一些Map和Reduce的函数接口,在运行期它会构建对应MapTask和ReduceTask,所以我们知道一个Job是由一个或多个MapTask,以及0个或1个ReduceTask组成。而对于MapTask,它是根据输入的数据文件的的逻辑分片(InputSplit)而定的,通常有多少个分片就会有多少个MapTask;而对于ReduceTask,它会根据我们编写的MapReduce程序配置的个数来运行。 有了这些信息,我们能够预想到,在Job运行过程中,无非也需要维护与这些Job/Task相关的一些状态信息,通过一定的调度策略来管理Job/Task的运行。这里,我们主要关注JobTracker端的一些