Apache Spark是一个开源的通用集群计算系统，它提供了High-level编程API，支持Scala、Java和Python三种编程语言。Spark内核使用Scala语言编写，通过基于Scala的函数式编程特性，在不同的计算层面进行抽象，代码设计非常优秀。 RDD抽象 RDD（Resilient Distributed Datasets），弹性分布式数据集，它是对分布式数据集的一种内存抽象，通过受限的共享内存方式来提供容错性，同时这种内存模型使得计算比传统的数据流模型要高效。RDD具有5个重要的特性，如下图所示： 上图展示了2个RDD进行JOIN操作，体现了RDD所具备的5个主要特性，如下所示： 一组分区 计算每一个数据分片的函数 RDD上的一组依赖 可选，对于键值对RDD，有一个Partitioner（通常是HashPartitioner） 可选，一组Preferred location信息（例如，HDFS文件的Block所在location信息） 有了上述特性，能够非常好地通过RDD来表达分布式数据集，并作为构建DAG图的基础：首先抽象一次分布式计算任务的逻辑表示，最终将任务在实际的物理计算环境中进行处理执行。 计算抽象 在描述Spark中的计算抽象，我们首先需
按作者浏览文章: Yanjun
Spark RPC通信层设计原理分析
Spark将RPC通信层设计的非常巧妙，融合了各种设计/架构模式，将一个分布式集群系统的通信层细节完全屏蔽，这样在上层的计算框架的设计中能够获得很好的灵活性。同时，如果上层想要增加各种新的特性，或者对来自不同企业或组织的程序员贡献的特性，也能够很容易地增加进来，可以避开复杂的通信层而将注意力集中在上层计算框架的处理和优化上，入手难度非常小。另外，对上层计算框架中的各个核心组件的开发、功能增强，以及Bug修复等都会变得更加容易。 Spark RPC层设计概览 Spark RPC层是基于优秀的网络通信框架Netty设计开发的，同时获得了Netty所具有的网络通信的可靠性和高效性。我们先把Spark中与RPC相关的一些类的关系梳理一下，为了能够更直观地表达RPC的设计，我们先从类的设计来看，如下图所示： 通过上图，可以清晰地将RPC设计分离出来，能够对RPC层有一个整体的印象。了解Spark RPC层的几个核心的概念（我们通过Spark源码中对应的类名来标识），能够更好地理解设计： RpcEndpoint RpcEndpoint定义了RPC通信过程中的通信端对象，除了具有管理一个RpcEndp
Apache Flink：特性、概念、组件栈、架构及原理分析
Apache Flink是一个面向分布式数据流处理和批量数据处理的开源计算平台，它能够基于同一个Flink运行时（Flink Runtime），提供支持流处理和批处理两种类型应用的功能。现有的开源计算方案，会把流处理和批处理作为两种不同的应用类型，因为他们它们所提供的SLA是完全不相同的：流处理一般需要支持低延迟、Exactly-once保证，而批处理需要支持高吞吐、高效处理，所以在实现的时候通常是分别给出两套实现方法，或者通过一个独立的开源框架来实现其中每一种处理方案。例如，实现批处理的开源方案有MapReduce、Tez、Crunch、Spark，实现流处理的开源方案有Samza、Storm。 Flink在实现流处理和批处理时，与传统的一些方案完全不同，它从另一个视角看待流处理和批处理，将二者统一起来：Flink是完全支持流处理，也就是说作为流处理看待时输入数据流是无界的；批处理被作为一种特殊的流处理，只是它的输入数据流被定义为有界的。基于同一个Flink运行时（Flink Runtime），分别提供了流处理和批处理API，而这两种API也是实现上层面向流处理、批处理类型应用框架的基础。 基本
Flume日志收集分层架构应用实践
Flume作为一个日志收集工具，非常轻量级，基于一个个Flume Agent，能够构建一个很复杂很强大的日志收集系统，它的灵活性和优势，主要体现在如下几点： 模块化设计：在其Flume Agent内部可以定义三种组件：Source、Channel、Sink 组合式设计：可以在Flume Agent中根据业务需要组合Source、Channel、Sink三种组件，构建相对复杂的日志流管道 插件式设计：可以通过配置文件来编排收集日志管道的流程，减少对Flume代码的侵入性 可扩展性：我们可以根据自己业务的需要来定制实现某些组件（Source、Channel、Sink） 支持集成各种主流系统和框架：像Hadoop、HBase、Hive、Kafka、ElasticSearch、Thrift、Avro等，都能够很好的和Flume集成 高级特性：Failover、Load balancing、Interceptor等 有关Flume的相关内容，可以参考官网文档，或者通过阅读我之前写的文章《Flume(NG)架构设计要点及配置实践》来快速了解。 为什么要对Flume日志收集系统进行分层设计 基于Flume设计实现分层日志收集系统，到底有什么好处呢？我们可以先看一下，如果不分层，会带来哪些问题： 如果需
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通过端口号来唯一标识，一个端口号对应一个Worker进程（如果该Worker进程被启动）。 ZooKeeper 用来协调Nimb
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, address, defaultConf); } }); ... ... JvmContext context
基于协同过滤的推荐方法
协同过滤（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(LaunchTaskAction action) { if (action.getTask().isMapTask()) { mapLauncher.addToTaskQueue(action);
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端的主要组件，如下图所示： 为了了解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的方法处理，每次交换一个中心点和非中心点，然后执行将非中心点指派到最近的中心点，计算得到的SAD值越小，则聚类质量越好，如此不断地迭代，直到找到一个最好
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值的一个簇，二分这个簇以后得到的2个子簇，选择2个子簇的总SSE最小的划分方法，这样能够保证每次二分得到的2个簇是比较优的（也可能是最优的
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算法的终止条件：质心在每一轮迭代中会发生变化，然后需要重新将非质心点指派给最近的质心而形成新的簇，如果只有很少的一部分点在迭代过程中，还在改变簇（如，更新一次质心，有些点从一个簇移动到另一