StarRocks 是新一代的 MPP 数据库，它具有很高的查询性能，能够支持各种场景的数据查询分析使用，主要包含如下几个场景：OLAP多维分析实时数据分析高并发查询统一分析关于 StarRocks 更详细的介绍，可以查看官方文档，非常详细（见后面参考链接）。下面，我们基于开源的 StarRocks 社区版，版本是 2.1.5 来进行集群的安装配置。StarRocks 集群的部署模式，采用 FE 与 BE 分离的模式。集群部署规划1 基础环境和软件软件版本操作系统CentOS-7.8StarRocks2.1.5JDKjdk1.8.0_212MySQL Client5.7.372 集群主机规划IP 地址主机名称角色备注信息172.168.0.1VM-0-1-centosFEFE Master172.168.0.2VM-0-2-centosFEFE OBSERVER172.168.0.3VM-0-3-centosFEFE FOLLOWER172.168.0.4VM-0-4-centosBEBE172.168.0.5VM-0-5-centosBEBE172.168.0.6VM-0-6-centosBEBE172.168.0.7VM-0-7-centosBEBE172.168.0.8VM-0-8-centosBEBE172.168.0.9VM-0-9-centosBEBE3 集群主机目录规 ...继续阅读 (94)

Apache Hudi是一个Data Lakes的开源方案，Hudi是Hadoop Updates and Incrementals的简写，它是由Uber开发并开源的Data Lakes解决方案。Hudi具有如下基本特性/能力：Hudi能够摄入（Ingest）和管理（Manage）基于HDFS之上的大型分析数据集，主要目的是高效的减少入库延时。Hudi基于Spark来对HDFS上的数据进行更新、插入、删除等。Hudi在HDFS数据集上提供如下流原语：插入更新（如何改变数据集）；增量拉取（如何获取变更的数据）。Hudi可以对HDFS上的parquet格式数据进行插入/更新操作。Hudi通过自定义InputFormat与Hadoop生态系统（Spark、Hive、Parquet）集成。Hudi通过Savepoint来实现数据恢复。目前，Hudi支持Spark 2.x版本，建议使用2.4.4+版本的Spark。基本架构与Kudu相比，Kudu是一个支持OLTP workload的数据存储系统，而Hudi的设计目标是基于Hadoop兼容的文件系统（如HDFS、S3等），重度依赖Spark的数据处理能力来实现增量处理和丰富的查询能力，Hudi支持Incremental Pulling而Kudu不支持。Hudi能够整合Batch和Streaming处理的能力，这是通过利用Spark自身支持的基 ...继续阅读 (80)

生成最佳执行计划，过程比较复杂，我们分成两篇来详细分析：上篇和下篇，本文为上篇。生成最佳执行计划是一个递归计算的过程：正向从DataSinkNode开始直到DataSourceNode，分别计算每个OptimizerNode的最佳计划，然后反向逐步将整个OptimizerNode DAG图转换为PlanNode DAG图，得到一个最优计划。其中，PlanNode的类继承结构，如下图所示：通过与OptimizerNode对应的节点结构类图对比，PlanNode更加抽象了一个层次，更关注Operator之间的数据交换策略。其中，生成最佳执行计划的过程，可以在Optimizer类中看到，如下代码所示：// the final step is now to generate the actual plan alternatives ListbestPlan = rootNode.getAlternativePlans(this.costEstimator);这是一个递归的过程，每个OptimizerNode都会通过获取到其孩子节点的最佳执行计划，从而递归地处理，从OptimizerNode DAG转换为PlanNode DAG。后面，我们会对SourcePlanNode、SinkPlanNode、SingleInputPlanNode、DualInputPlanNode创建的处理过程进行详 ...继续阅读 (73)

这里，我们主要简单介绍在卷积神经网络发展过程中，一些经常用的改进模型，主要包括LeNet-5、AlexNet、VGGNet、GoogLeNet、ResNet、DenseNet、ZFNet这7个模型。本文不会非常深入讲解各个CNN模型，而是希望能够快速了解到各个模型起源，基本结构是什么样子，以及其它模型相比有什么明显的不同。LeNet-5LeNet-5是第一个由Yann LeCun提出的卷积神经网络，它也是最基础的一个卷积神经网络，网络结构可以参考论文《Gradient-Based Learning Applied To Document Recognition》，如下图所示：LeNet-5是一个8层CNN网络（包含输入层），其中包含卷积层块和全连接层块两个部分。卷积层用来识别图像里的空间模式，如线条和物体局部，之后的最大池化层则用来降低卷积层对位置的敏感性，卷积层块由两个这样的基本单位重复堆叠构成。当卷积层块的输出传入全连接层块时，全连接层块会将小批量中每个样本变平（Flatten）。AlexNetAexNet模型的名字来源于论文第一作者Alex Krizhevsky的名字，使用了8层卷积神经网络，并以很大的优势赢得了ImageNet 2012图像识别挑战赛。AlexNet网络模型，可以参考论文《Imagenet Classification With Deep Convoluti ...继续阅读 (77)

卷积神经网络（Convolutional Neural Networks，CNN）是由纽约大学的Yann Lecun于1998年提出的，其本质是一个多层感知机，它是一类包含卷积计算且具有深度结构的前馈神经网络（Feedforward Neural Networks），是深度学习（Deep Learning）的代表算法之一。卷积神经网络是一种特殊的多层神经网络，像其它的神经网络一样，卷积神经网络也使用一种反向传播算法来进行训练，不同之处在于网络的结构。卷积神经网络（CNN）具有一些传统技术所没有的优点：良好的容错能力、并行处理能力和自学习能力，可处理环境信息复杂，背景知识不清楚，推理规则不明确情况下的问题；它允许样本有较大的缺损、畸变，运行速度快，自适应性能好，具有较高的分辨率；它是通过结构重组和减少权值将特征抽取功能融合进多层感知器，省略识别前复杂的图像特征抽取过程。CNN基本特征下面，我们根据网上大家分享的有关卷积神经网络（CNN）的内容，梳理总结CNN所具有的一些特征，如下所示：具有多层层次网络结构卷积神经网络（CNN）被认为是第一个真正成功的、采用多层层次结构网络的、具有鲁棒性的深度学习方法，它通过挖掘数据在空间上的相关性，来减少网络中的可训练参数的数量，从而改进了前向传播网络的反向传播算法效率。在CNN中，图像中的小块区域（也叫做“局部感知区域”）被当做层次结构中的底层的输 ...继续阅读 (63)

使用TensorFlow官方提供了一个例子，基于MNIST数据集，实现一个图片分类的应用，本文是基于TensorFlow 2.0.0版本来学习和试验的。MNIST数据集是一个非常出名的手写体数字识别数据集，它包含了60000张图片作为训练集，10000张图片作为测试集，每张图片中的手写体数字是0~9中的一个，图片是28×28像素大小，并且每个数字都是位于图片的正中间的。使用TensorFlow对MNIST数据集进行分类，整个实现对应的完整的Python代码，如下所示：from __future__ import absolute_import, division, print_function, unicode_literals import tensorflow as tf # 下载 MNIST 数据集 mnist = tf.keras.datasets.mnist (x_train, y_train), (x_test, y_test) = mnist.load_data() x_train, x_test = x_train / 255.0, x_test / 255.0 # 创建 tf.keras.Sequential 模型 model = tf.keras.models.Sequential([ tf.keras.layers.Flatten(inp ...继续阅读 (71)

OptimizerNode DAG，是基于程序Plan中创建好的Operator DAG，并以OptimizerNode作为DAG图节点构建完成的。所以，我们先看一下组成OptimizerNode DAG，都有哪些类型的OptimizerNode，如下图所示：通过上面类图，可以看到其中主要有SingleInputNode、TwoInputNode、DataSourceNode、DataSinkNode等这几大类，根据各个OptimizerNode具体实现类名称，可以对应到具体操作功能的Operator实现类。比如，MapNode对应map操作，JoinNode对应join操作，等等。OptimizerNode结构下面看一下OptimizerNode的数据结构，了解它都抽象了哪些内容，如下图所示：上图中的结构是每个OptimizerNode都包含的内容，具体到每个OptimizerNode实现，都会根据Operator的属性及行为，补充一些其他的信息。上面的cachedPlans是一个基于PlanNode的DAG的列表，在生成OptimizerNode过程中并没有用到，而是基于OptimizerNode DAG生成最佳执行计划的时候，通过PlanNode DAG来表示的，后续文章我们会单独详细分析，这里先跳过。下面将对OptimizerNode节点结构包含的一些重要属性信息进行说明， ...继续阅读 (90)

我们这里所说的优化执行计划，是指从生成Plan之后，一直到生成JobGraph之前这一期间对Plan进行各种转换、增强、优化处理。从用户编程的Flink批处理程序开始，一直到生成JobGraph的整个过程，这个过程中，会创建很多数据结构来完成执行计划的优化工作，而且，通过分层增强与优化的方式，有利于我们对每层进行了哪些处理有一个很好地理解。我们从整理流程上来看，都有哪些核心的处理，然后会对每一步进行概要说明，从宏观上理解每一步都做了哪些操作。批处理执行计划生成及优化的主要流程，如下图所示：根据上图，我们将Flink批处理执行计划生成的流程，分为如下几个阶段进行说明。构建用户编程Operator DAG把用户程序中设置的各种DataSource、Transformation、DataSink及其配置信息进行收集，以最原始的形态保存在ExecutionEnvironment上线文环境对象中，当然也跟踪了从DataSource到DataSink之间的操作顺序，最后通过ExecutionEnvironment对象，基于DataSink就可以从后向前遍历，直接链接到DataSource。在上面的处理过程中，涉及到3个比较重要的内容：一是用户根据Flink定义的各种Transformation接口实现并传入的处理函数，也包括Flink为了完成处理内部直接创建的函数对象；二是创建位于org.ap ...继续阅读 (89)

我们先通过一个简单的批量处理WordCount的程序，来了解一下，在编程API层面的基本使用方法，以及Flink是如何基于Pipeline风格的API实现一个完整的应用程序的，还有就是每一步操作的底层，都是如何进行转换的。WordCount程序代码，示例如下所示：final MultipleParameterTool params = MultipleParameterTool.fromArgs(args); final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setGlobalJobParameters(params); // get input data DataSettext = env.readTextFile(input); DataSet> counts = // split up the lines in pairs (2-tuples) containing: (word,1) text.flatMap(new Tokenizer()) // grou ...继续阅读 (68)

默认配置情况下，Spark在Shuffle过程中会使用SortShuffleManager来管理Shuffle过程中需要的基本组件，以及对RDD各个Partition数据的计算。我们可以在Driver和Executor对应的SparkEnv对象创建过程中看到对应的配置，如下代码所示：// Let the user specify short names for shuffle managers val shortShuffleMgrNames = Map( "sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName, "tungsten-sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName) val shuffleMgrName = conf.get("spark.shuffle.manager", "sort") val shuffleMgrClass = shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName) val shuffleManager ...继续阅读 (64)

Apache Spark中，对Block的查询、存储管理，是通过唯一的Block ID来进行区分的。所以，了解Block ID的生成规则，能够帮助我们了解Block查询、存储过程中是如何定位Block以及如何处理互斥存储/读取同一个Block的。可以想到，同一个Spark Application，以及多个运行的Application之间，对应的Block都具有唯一的ID，通过代码可以看到，BlockID包括：RDDBlockId、ShuffleBlockId、ShuffleDataBlockId、ShuffleIndexBlockId、BroadcastBlockId、TaskResultBlockId、TempLocalBlockId、TempShuffleBlockId这8种ID，可以详见如下代码定义：@DeveloperApi case class RDDBlockId(rddId: Int, splitIndex: Int) extends BlockId { override def name: String = "rdd_" + rddId + "_" + splitIndex } // Format of the shuffle block ids (including data and index) should be kept in sync with // ...继续阅读 (64)

Docker集群管理和编排的特性是通过SwarmKit进行构建的， 其中Swarm mode是Docker Engine内置支持的一种默认实现。Docker 1.12以及更新的版本，都支持Swarm mode，我们可以基于Docker Engine来构建Swarm集群，然后就可以将我们的应用服务（Application Service）部署到Swarm集群中。创建Swarm集群的方式很简单，先初始化一个Swarm集群，然后将其他的Node加入到该集群即可。本文主要基于Docker Swarm官网文档，学习总结。基本特性Docker Swarm具有如下基本特性：集群管理集成进Docker Engine使用内置的集群管理功能，我们可以直接通过Docker CLI命令来创建Swarm集群，然后去部署应用服务，而不再需要其它外部的软件来创建和管理一个Swarm集群。去中心化设计Swarm集群中包含Manager和Worker两类Node，我们可以直接基于Docker Engine来部署任何类型的Node。而且，在Swarm集群运行期间，我们既可以对其作出任何改变，实现对集群的扩容和缩容等，如添加Manager Node，如删除Worker Node，而做这些操作不需要暂停或重启当前的Swarm集群服务。声明式服务模型（Declarative Service Model）在我们实现的应用栈中 ...继续阅读 (83)

Docker是一个开源的应用容器引擎，开发人员可以非常容易地打包已经开发好的应用，同时将应用相关的依赖包也打包到这样一个可移植的容器中，然后发布到任意的Linux主机系统上。Docker是基于Linux Container（LXC）技术实现的一个轻量级虚拟化解决方案，用户可以直接使用容器（Container），来构建自己的应用程序，应用开发人员无需将注意力集中在容器的管理上。Docker的目标是“Build, Ship and Run Any App, Anywhere”，这说明了使用Docker能够实现应用运行的可移植性、便捷性，对开发人员非常友好，只要你的应用是基于Docker进行构建和部署的，在任何时候任何支持Docker的Linux发行版操作系统上都可以运行你的应用程序。Docker是基于Go语言开发的， 代码开源，可以在Github上查看对应的源码：https://github.com/docker/docker.git。基本构架Docker基于Client-Server架构，Docker daemon是服务端，Docker client是客户端。Docker的基本架构，如下图所示：上图中，除了展现了Docker的Client、Server、Containers、Images、Registry之间的关系，我们主要说明Docker daemon和Docker client， ...继续阅读 (74)

在使用Spring Cloud Netflix构建微服务之前，我们先了解一下Spring Cloud集成的Netflix OSS的基础组件Eureka，对于Netflix的其他微服务组件，像Hystrix、Zuul、Ribbon等等本文暂不涉及，感兴趣可以参考官网文档。这里，我们用最基础的Eureka来构建一个最基础的微服务应用，来演示如何构建微服务，了解微服务的基本特点。EurekaEureka是Netflix开源的一个微服务注册组件，提供服务发现特性，它是一个基于REST的服务，主要具有如下功能：支持服务注册和发现具有Load Balance和Failover的功能在进行服务调用过程中，无需知道目标服务的主机（IP）和端口，只要知道服务名就可以实现调用通过Netfix在Github上的文档，我们看一下Eureka的基本架构，如下图所示：Eureka主要包含如下两个核心组件：Eureka ServerEureka Server是服务注册的服务端组件，负责管理Eureka Client注册的服务，提供服务发现的功能。它支持集群模式部署，集群部署模式中，多个Eureka Server之间会同步服务注册数据，能够保证某一个Eureka Server因为故障挂掉，仍能对外提供注册服务的能力。因为最初在Netflix，Eureka主要用在AWS Cloud上，用作定位服务、Load Bal ...继续阅读 (67)

Spark的内存使用，大体上可以分为两类：Execution内存和Storage内存。在Spark 1.5版本之前，内存管理使用的是StaticMemoryManager，该内存管理模型最大的特点就是，可以为Execution内存区与Storage内存区配置一个静态的boundary，这种方式实现起来比较简单，但是存在一些问题：没有一个合理的默认值能够适应不同计算场景下的Workload内存调优困难，需要对Spark内部原理非常熟悉才能做好对不需要Cache的Application的计算场景，只能使用很少一部分内存为了克服上述提到的问题，尽量提高Spark计算的通用性，降低内存调优难度，减少OOM导致的失败问题，从Spark 1.6版本开始，新增了UnifiedMemoryManager（统一内存管理）内存管理模型的实现。UnifiedMemoryManager依赖的一些组件类及其关系，如下类图所示：从上图可以看出，最直接最核心的就是StorageMemoryPool 和ExecutionMemoryPool，它们实现了动态内存池（Memory Pool）的功能，能够动态调整Storage内存区与Execution内存区之间的Soft boundary，使内存管理更加灵活。下面我们从内存布局和内存控制两个方面，来分析UnifiedMemoryManager内存管理模型。内存布局Un ...继续阅读 (76)

Apache Beam是一个开源的数据处理编程库，由Google共享给Apache的项目，前不久刚刚成为Apache TLP项目。它提供了一个高级的、统一的编程模型，允许我们通过构建Pipeline的方式实现批量、流数据处理，并且构建好的Pipeline能够运行在底层不同的执行引擎上。刚刚接触该开源项目时，我的第一感觉就是：在编程API的设计上，数据集及其操作的抽象有点类似Apache Crunch（MapReduce Pipeline编程库）项目；而在支持统一数据处理模型上，能够让人想到Apache Flink项目。如果深入了解Apache Beam，你会发现未来Apache Beam很可能成为数据处理领域唯一一个能够将不同的数据应用统一起来的编程库。Apache Beam架构概览Apache Beam目前最新版本为0.5.0-SNAPSHOT，最新的Release版本为0.4.0，很多特性还在开发中。在网上找到一个由Andrew Psaltis在2016年6月份演讲的《Apache Beam: The Case for Unifying Streaming API’s》，引用了其中一个Apache Beam的架构图，如下图所示：上图中，我们可以看到，Apache Beam核心的主要有两层：Pipeline构建层在Pipeline构建层，针对不同的编程语言，构建一组用 ...继续阅读 (75)

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中的计算抽象，我们首先需要了解如下几个概念：Application用户编写的Spark程序，完成一个计算任务的处理。它是由一个Driver程序和一组运行于S ...继续阅读 (76)

Spark将RPC通信层设计的非常巧妙，融合了各种设计/架构模式，将一个分布式集群系统的通信层细节完全屏蔽，这样在上层的计算框架的设计中能够获得很好的灵活性。同时，如果上层想要增加各种新的特性，或者对来自不同企业或组织的程序员贡献的特性，也能够很容易地增加进来，可以避开复杂的通信层而将注意力集中在上层计算框架的处理和优化上，入手难度非常小。另外，对上层计算框架中的各个核心组件的开发、功能增强，以及Bug修复等都会变得更加容易。Spark RPC层设计概览Spark RPC层是基于优秀的网络通信框架Netty设计开发的，同时获得了Netty所具有的网络通信的可靠性和高效性。我们先把Spark中与RPC相关的一些类的关系梳理一下，为了能够更直观地表达RPC的设计，我们先从类的设计来看，如下图所示：通过上图，可以清晰地将RPC设计分离出来，能够对RPC层有一个整体的印象。了解Spark RPC层的几个核心的概念（我们通过Spark源码中对应的类名来标识），能够更好地理解设计：RpcEndpointRpcEndpoint定义了RPC通信过程中的通信端对象，除了具有管理一个RpcEndpoint生命周期的操作（constructor -> onStart -> receive* -> onStop），并给出了通信过程中一个RpcEndpoint所具有的基于事件驱动的行为（连 ...继续阅读 (69)

Apache Flink是一个面向分布式数据流处理和批量数据处理的开源计算平台，它能够基于同一个Flink运行时（Flink Runtime），提供支持流处理和批处理两种类型应用的功能。现有的开源计算方案，会把流处理和批处理作为两种不同的应用类型，因为他们它们所提供的SLA是完全不相同的：流处理一般需要支持低延迟、Exactly-once保证，而批处理需要支持高吞吐、高效处理，所以在实现的时候通常是分别给出两套实现方法，或者通过一个独立的开源框架来实现其中每一种处理方案。例如，实现批处理的开源方案有MapReduce、Tez、Crunch、Spark，实现流处理的开源方案有Samza、Storm。Flink在实现流处理和批处理时，与传统的一些方案完全不同，它从另一个视角看待流处理和批处理，将二者统一起来：Flink是完全支持流处理，也就是说作为流处理看待时输入数据流是无界的；批处理被作为一种特殊的流处理，只是它的输入数据流被定义为有界的。基于同一个Flink运行时（Flink Runtime），分别提供了流处理和批处理API，而这两种API也是实现上层面向流处理、批处理类型应用框架的基础。基本特性关于Flink所支持的特性，我这里只是通过分类的方式简单做一下梳理，涉及到具体的一些概念及其原理会在后面的部分做详细说明。流处理特性支持高吞吐、低延迟、高性能的流处理支持带有事件时间的窗口 ...继续阅读 (88)

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设计实现分层日志收集系统，到底有什么好处呢？我们可以先看一下，如果不分层，会带来哪些问题：如果需要通过Kafka去缓冲上游基于Flume收集而构建的日志流，对于数据平台内部服务器产生的数据还好，但是如果日志数据是 ...继续阅读 (81)

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

我们基于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, ...继续阅读 (100)

协同过滤（Collaborative Filtering， CF）是推荐系统广泛使用的一种技术，它主要通过考虑用户（User）与用户之间、物品（Item）与物品之间的相似度（Similarity），来向用户推荐物品，常被用在电商网站中。其中，在推荐系统中最常使用的协同过滤方法，有如下4种：基于用户的协同过滤推荐基于物品的协同过滤推荐基于模型的协同过滤推荐混合协同过滤推荐上面4种方法中，基于用户的协同过滤推荐、基于物品的协同过滤推荐都是基于内存的协同过滤推荐，一般在数据量较小的应用场景下，可以直接在线使用的实时推荐方法；基于模型的协同过滤推荐一般用于离线计算，它采用机器学习的方法，一般首相将用户偏好行为数据分成2个数据集（有时可能会将数据集分成k个子集，采用交叉验证的方式来提高模型精度），一个为训练集，一个为测试集，使用训练集数据来训练出推荐模型，然后使用测试集数据来评估模型的精度，当满足特定精度时，可以将得到的推荐模型应用于实际线上环境；混合协同过滤推荐，是综合基于内存的协同过滤（基于用户的协同过滤推荐、基于物品的协同过滤推荐）、基于模型的协同过滤推荐这两类方法，克服每一种协同过滤方法本身的缺点，比如基于内存的协同过滤推荐方法不适用于大量数据的推荐应用场景，而基于模型的协同过滤推荐方法由于所基于的数据多是大量历史记录，训练模型时间较长，它不适用于在线推荐场景，综合各种协同过滤推荐方 ...继续阅读 (82)

我们基于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.addToTas ...继续阅读 (95)

我们基于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中各个组件都负责处理哪些工作，我们通过下表来简要地说明各个组件的功能，如下表所示：组件名称组件功能localFs: FileSys ...继续阅读 (84)

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值越小，则聚类质量越好，如此不断地迭代，直到找到一个最好的划分。在维基百科上，它给出的基于PAM方法计算聚类的过程，描述如下：从待聚类的数据点集中随机选择k个点，作为初始中心点；将待聚类的数据点集中的点，指派到最近的中心点；进入迭代，直到聚类的质量满足指定的阈值（可以通过计算SAD） ...继续阅读 (69)

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个簇是比较优的（也可能是最优的），也就是这2个簇的划分可能是局部最优的，取决于试验的次数。Bisecting k-means聚类算法的具体执行过程，描述如下所示：初始时，将待聚类数据集D作为一个簇C0，即C={C0}，输 ...继续阅读 (88)

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算法的终止条件：质心在每一轮迭代中会发生变化，然后需要重新将非质心点指派给最近的质心而形成新的簇，如果只有很少的一部分点在迭代过程中，还在改变簇（如，更新一次质心，有些点从一个簇移动到另一个簇），那么满足这样一个收敛条件，可以提前结束迭代过程。k-means算法的框架是：首先随机选择k个初始质心点，然后执行聚类处理迭代，不断更新质心，直到满足 ...继续阅读 (83)

ElasticSearch是基于全文搜索引擎库Lucene构建的分布式搜索引擎，我们可以直接使用ElasticSearch实现分布式搜索系统的搭建与使用，都知道，Lucene只是一个搜索框架，它提供了搜索引擎操作的基本API，如果要实现一个能够使用的搜索引擎系统，还需要自己基于Lucene的API去实现，工作量很大，而且还需要很好地掌握Lucene的底层实现原理。ElasticSearch是一个完整的分布式搜索引擎系统，它的一些基本特性包括如下：全文检索提供插件机制，可以共享重用插件的功能分布式文件存储分布式实时索引和搜索实时统计分析可以横向扩展，支持大规模数据的搜索简单易用的RESTful API基于Replication实现了数据的高可用特性与其他系统的集成支持结构化和非结构化数据灵活的Schema设计（Mappings）支持多编程语言客户端我个人感觉，ElasticSearch尽量屏蔽底层Lucene相关的技术细节，让你根本无从感觉底层Lucene相关的内容，这样你可以省去了了解Lucene 的成本，学习曲线比较平缓，不像Solr，如果想要构造负责的查询（Query），还是要对Lucene有所了解的。另外，在分布式设计方面，ElasticSearch更轻量一些，用起来更简单，而使用Solr的分布式分片功能需要使用SolrCloud，它基于ZooKeeper来实现配置管理，以及 ...继续阅读 (306)

我们基于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, short responseId) throws IOE ...继续阅读 (75)