Storm实时数据处理

2.6　统计与持久化日志统计信息 / 53
2.7　为日志流集群创建集成测试 / 55
2.8　创建日志分析面板 / 59
第3章　使用Trident计算单词重要度 / 71
3.1　简介 / 71
3.2　使用Twitter过滤器创建URL流 / 71
3.3　从文件中获取整洁的词流 / 76
3.4　计算每个单词的相对重要度 / 81
第4章　分布式远程过程调用 / 85
4.1　简介 / 85
4.2　通过DPRC实现所需处理流程 / 85
4.3　对Trident Topology进行集成测试 / 90
4.4　实现滚动窗口Topology / 95
4.5　在集成测试中模拟时间 / 98
第5章　在不同语言中实现Topology / 100
5.1　简介 / 100
5.2　在Qt中实现多语言协议 / 100
5.3　在Qt中实现SplitSentence Bolt / 105
5.4　在Ruby中实现计数 Bolt / 108
5.5　在Clojure中实现单词计数Topology / 109
第6章　Storm与Hadoop集成 / 113
6.1　简介 / 113
6.2　在Hadoop中实现TF-IDF算法 / 115
6.3　持久化来自Storm的文件 / 121
6.4　集成批处理与实时视图 / 122
第7章　实时机器学习 / 127
7.1　简介 / 127
7.2　实现事务性Topology / 129
7.3　在R中创建随机森林分类模型 / 134
7.4　基于随机森林的事务流业务分类 / 143
7.5　在R中创建关联规则模型 / 149
7.6　创建推荐引擎 / 152
7.7　实时在线机器学习 / 157
第8章　持续交付 / 162
8.1　简介 / 162
8.2　搭建CI服务器 / 162
8.3　搭建系统环境 / 164
8.4　定义交付流水线 / 166
8.5　实现自动化验收测试 / 170
第9章　在AWS上部署Storm / 177
9.1　简介 / 177
9.2　使用Pallet在AWS上部署Storm / 177
9.3　 搭建虚拟私有云 / 181
9.4　使用Vagrant在虚拟私有云上部署Storm / 189

.　　阅读前提
　　本书采用Ubuntu或Debian操作系统作为基本的开发环境。第1章会介绍如何配置其余必备的工具。如果你不能使用Ubuntu作为开发用的操作系统，任何基于*Nix的操作系统都是可以的，因为书中所有示例都以bash命令行界面为基础构建。
　　读者对象
　　如果你想学习实时处理技术，或想了解通过Storm实现实时处理的方法，那么本书非常适合你。本书假定你是一名Java开发者，要是你还具备Clojure、C++和Ruby的开发经验，那将是锦上添花的事情，不过那并不是必需的。如果你还了解一些Hadoop或类似的技术就更好了。
　　本书版式约定
　　在本书中，读者会发现针对不同信息类型的文本样式。下面是这些样式的示例和解释。
　　代码段版式如下所示：
　　所有命令行输入和输出如下所示：
　　下载本书的示例代码
　　访问http://www.packtpub.com/，可以下载本书及你所购买的所有Packt图书的示例代码。如果你是从其他地方购买的本书英文版，可以访问http://www.packtpub.com/support并注册，以便通过电子邮件取得示例文件。
　　作者通过Bitbucket账户来维护开源版本的代码：https://bitbucket.org/qanderson。
　　致谢
　　感谢Storm社区，正是他们的努力构建出了开源社区中如此出色的平台，当然，还要特别感谢Storm的核心作者—Nathan Marz。
　　我要感谢我的妻子和孩子对我长时间写作本书和其他相关项目的宽容，感谢这段时间他们给予我的帮助，我爱他们。此外，我还要感谢所有参与本书审校工作的朋友。
　　

.　　Step05最后，你还需要安装一个集成开发环境（IDE）：
　　自从Sun公司被Oracle收购以后，人们就一直在争论该使用哪种Java SDK。虽说作者理解OpenJDK的必要性，但书中的例子都是在Oracle JDK下测试通过的。总的来说，OpenJDK和Oracle JDK没有什么区别，只是Oracle JDK为了追求稳定而在功能方面相对滞后。
　　1.2.2解析
　　对于任何Java开发工作来说JDK都是必不可少的。GIT是近几年被广泛采用的开源分布式版本控制系统。稍后我们将简要介绍GIT。
　　Maven是一种广泛使用的“约定优于配置”的构建系统。Maven包含了许多有用的功能，比如项目对象模型（POM），POM能够让我们以有效的方式管理库、依赖和版本。Maven在互联网上拥有许多二进制仓库可供使用，所以我们可以直接用恰当的方式来维护二进制依赖项，然后打包部署Topology。
　　由于DevOps和持续交付的不断发展，Puppet 系统已经被广泛用于为Linux、其他一些操作系统和应用程序提供声明式服务器配置。Puppet能够对服务器的状态和部署环境进行编程，这一点非常重要，因为这样就可以通过GIT版本控制系统管理服务器状态，而且对于服务器的修改都可以安全删除。这么做好处很多，包括可确定的服务器平均恢复时间（MTTR）和审计跟踪，说白了就是让系统更加稳定。这对实现持续交付来说至关重要。
　　Vagrant是一个在开发过程中非常有用的工具，它能自动配置VirtualBox虚拟机，这对Storm处理系统这样的以集群为基础的技术来说尤为重要。为了测试一个集群，你必须构建一个真正的机群或配置多台虚拟机。Vagrant能准确无误地在本地配置虚拟机。
　　虚拟机在IT运维和系统开发过程中能够大显身手，但需要指出的是，虽然我们都知道在本地部署的虚拟机性能不太理想，但其可用性却完全依赖于可用的内存容量。处理能力不是需要考虑的主要问题，因为现代处理器的利用率普遍严重偏低，尽管处理Topology时利用率会高一些。建议你的计算机至少保留8GB的内存容量。
　　1.3分布式版本控制
　　传统版本控制系统都是集中式的。每个客户端都包含一份从当前版本签出的文件，而当前版本则取决于客户端使用的分支。所有历史版本都会存储在服务器上。这样的做法效果不错，不仅能让团队紧密协作，还能知道其他成员在做什么工作。
　　但集中式服务器存在一些较为明显的缺点，这让分布式版本控制系统的应用范围越来越广。首先，集中式服务器存在单点故障问题，如果服务器由于某种原因死机或无法访问，开发人员会因此难以继续工作。其次，如果服务器上的数据由于某种原因损坏或丢失，代码库的历史记录也就随之丢失。
　　基于以上两个原因，开源项目极大地推动了分布式版本控制的发展，但究其主要原因，还是分布式系统能帮助我们实现协作模型。开发人员可以在本地环境下用自己的方式进行工作，然后在方便的时候，一次性或多次将这些改动分布到一个或多个远程代码仓库。
　　除此之外，还有一个很明显的优势就是，由于每一个客户端都拥有代码仓库的完整镜像，因此会自然存在多个仓库备份。所以，如果任何一个客户端或服务器出现问题，都可以轻松恢复，实现数据还原。
　　实战
　　本书使用Git作为分布式版本控制系统。如果要建立一个代码仓库，你可以克隆已存在的代码仓库，或者通过初始化创建一个新的代码仓库。对于新创建的项目来说，应该通过初始化创建一个新的代码仓库。
　　Step01通过以下命令创建项目目录：
　　Step02为了确认我们之前的步骤是否有效，需要在代码仓库中添加一些文件：
　　使用vim或任何其他文本编辑器，简单增加一些描述性文字并按下插入键。完成后按下Esc键，输入“:wq”，并按下回车键保存。
　　Step03在提交之前，我们来看看代码仓库的状态：
　　该命令会显示出类似于以下内容的信息：
　　Step04你需要使用以下命令，手动添加所有的文件和目录到Git代码仓库中：
　　Step05然后提交：
　　Step06这时会打开vim编辑器并让你添加注释。
　　你可以在输入命令的同时，使用-m参数指定要提交的描述信息。
　　如果不把代码仓库存储在远程主机上，你将遇到和集中式存储一样的风险问题。所以你应该把代码仓库存储在远程主机上。www.github.com和www.bitbucket.org是两个不错的免费Git托管服务，这两个服务都可以通过加密保证你的知识产权不被公共用户访问。本书使用bitbucket.org作为示例。要使用该远程托管服务，只需在你的浏览器上访问该站点并注册一个账户。
　　注册成功后，点击菜单创建一个新的代码仓库，如图1-1所示。
　　图1-1点击菜单创建新代码仓库
　　填写图1-2所示内容来创建代码仓库：
　　图1-2创建代码仓库界面
　　代码仓库创建完成后，你需要将远程代码仓库添加到你的本地代码仓库中，然后向远程代码仓库提交修改。
　　你需要使用刚才注册的用户名替换上面命令行中的[user]。
　　稍后我们会介绍克隆代码仓库的方法以及如何通过版本控制进行工作。
　　1.4创建“Hello World”Topology
　　“Hello World” Topology和其他所有的“Hello World”应用程序一样，并没有什么实际用途，其目的在于说明一些基本概念。“Hello World” Topology结构将演示如何创建一个包含简单的Spout和Bolt的项目，如何构建项目，并在本地集群模式下运行项目。
　　1.4.1实战
　　Step01新建一个项目目录，并初始化你的Git代码仓库。
　　Step02然后，我们需要通过以下命令创建Maven项目文件：
　　Step03使用vim或任何其他文本编辑器，为“Hello World”项目添加基本的XML标签和项目元数据。
　　Step04然后，我们需要指定从哪一个Maven仓库获取依赖项。在pom.xml文件的项目标签内添加以下内容：
　　你可以使用.m2和settings.xml文件替换上面指定的这些仓库，虽然具体方法超出了本书的讨论范围，但对于开发团队来说依赖项管理非常重要，因此这种方法会变得十分有用。
　　Step05然后在项目标签中添加我们所需的依赖项：
　　Step06最后，我们需要为Maven添加build插件定义：
　　Step07当创建好POM文件后，通过“Esc + : + wq + 回车键”保存文件，并为Maven项目创建所需的目录结构：
　　Step08然后回到项目根目录，并通过以下命令创建Eclipse项目文件：
　　Eclipse的项目文件类似于.class文件，属于生成的工件，在提交到Git代码仓库的时候不应该包含此文件，特别是因为它们包含了客户机特定的路径。
　　Step09打开你的Eclipse环境并把刚才生成的项目文件导入到工作区中，如图1-3所示。
　　图1-3导入项目到工作区
　　Step10接下来就可以创建你的第一个Spout了，在storm.cookbook包中创建名为HelloWorldSpout的类，并继承BaseRichSpout。Eclipse会为你生成一个默认的spouts方法。Spout会生成一个随机数生成的Tuple。添加以下成员变量并构造对象：
　　Step11对象构造好后，Storm集群将会打开Spout，为open方法添加以下实现代码：
　　Step12Storm集群会重复调用nextTuple方法，该方法包含了整个Spout的逻辑。为该方法添加以下实现代码：
　　Step13最后，你需要通过declareOutputFields方法，告诉Storm集群Spout发送了哪些字段：
　　Step14导入类所需的包和类后，就可以开始创建HelloWorldBolt了。该类用于读取已产生的Tuple并实现必要的统计逻辑。在storm.cookbook包中创建一个继承BaseRichBolt的新类。为Bolt声明一个私有成员变量并为execute方法添加以下实现代码：
　　Step15最后，将所有代码逻辑组织起来，声明Storm Topology。在同一个包中创建名为HelloWorldTopology的主类，并添加以下主函数实现：
　　这样就搭建好了Topology，并根据传递给main方法的参数，将其提交到本地或远程的Storm集群上。
　　Step16解决所有的编译问题后，你就能在项目根目录下，通过以下命令启动集群了：
　　1.4.2解析
　　图1-4描述了“Hello World” Topology：
　　图1-4“Hello World” Topology
　　Spout最终会发送一个流（Stream），包含下面两条语句中的任意一条：
　　Other Random Word
　　Hello World
　　Spout根据随机概率决定要发送哪一条语句，具体原理是：先在构造函数中生成一个随机数，作为原始变量的值，接着生成随后的随机数，并和原始成员变量的值进行比较。若比较结果一致，就发送“Hello World”，否则会发送另一条语句。
　　Bolt会简单对比和计算Hello World的数量。在我们当前的实现中，你会注意到一系列通过Bolt打印出来的增量。想要扩展Bolt的规模，只需通过更新以下语句就可以为Topology增加并行程度：
　　这里parallism_hint是关键参数，你可以根据情况把这个参数值调高。如果重新运行集群，你将能看到三条独立的计数信息交织地打印出来。
　　部署运行后若要调整集群的大小，可以通过Storm的GUI或CLI来修改并行程度。但无法在没有重新编译和部署JAR包的情况下，直接修改Topology的结构。有关命令行选项的信息，请通过以下链接查看wiki上的CLI文档：https://github.com/nathanmarz/storm/wiki/Command-line-client。
　　确保POM中指定的项目依赖项正确是很重要的。我们必须将Storm JAR文件的范围设定成provided，否则这些依赖项会被打包进你的JAR，这会导致在一个集群节点中，classpath下出现重复的类文件。而且Storm会检查这种classpath重复的问题。启动集群会因为部署中包含了Storm的文件而失败。
　　下载示例代码
　　访问http://www.packtpub.com可下载本书及你所购买的所有Packt图书的示例代码。如果你是从其他地方购买的本书英文版，可以访问http://www.packtpub.com/support并注册，以便通过电子邮件取得示例文件。
　　作者通过Bitbucket账户来维护开源版本的代码：https://bitbucket.org/qanderson。
　　1.5创建Storm集群——配置机器
　　本地模式下测试集群对调试和验证集群的基本功能逻辑很有帮助。但是，这并不代表你就能够了解集群在实际环境中运行的状况。此外，只有当系统已经在产品环境中运行时，开发工作才算真正完成。任何开发者都应该重视这一点，并且这也是整个DevOps实践的基础。无论采用什么方法，你都必须能够将代码可靠地部署到产品环境中。本节将展示如何直接通过版本控制创建和配置一个完整的集群。在此之前，需要事先说明一些有关创建和配置集群的基本原则：
　　我们需要时刻了解服务器的会话状态。在没有严格版本控制的情况下，不允许登录和修改服务器的设置或文件。
　　不能随意修改服务器及其独立存储卷中的状态信息，这才能确保服务器的恢复时间。
　　如果在交付过程中遇到了问题，就应该多尝试几次。这对软件开发和信息技术运维中的灾难恢复和集成工作来说，是个不小的负担。而只有恢复和集成工作可以自动化进行的时候，我们才能重复操作。
　　本书假定你使用的产品部署环境是具有可伸缩性的集群（基于Amazon Web Services（AWS）EC2）。因为只有具备可伸缩性的集群才能实现自动化配置。
　　我们会在后续章节介绍AWS集群上部署Storm Topology的方法，在此之前会在本节的开发环境中对此方法做一些基本的介绍。
　　1.5.1实战
　　我们按照下面的方法新建一个项目。
　　Step01按照以下目录结构，新建一个名为vagrant-storm-cluster的项目：
　　图1-5项目的目录结构
　　Step02用你最喜爱的编辑器，在项目根目录下创建一个名为Vagrantfile的文件。在文件里添加文件头和我们想创建的虚拟机的配置。文件中至少包含一个nimbus节点、两个supervisor节点和一个zookeeper节点：
　　这里要注意的是，由于我们不打算设计具备高可用性的集群，因此在开发环境只用了一个zookeeper节点。该集群的作用是在现实环境中测试你的Topology逻辑，并确定其稳定性。
　　Step03然后，就可以为每台机器创建虚拟机配置项了，Vagrant运行的时候会根据这些配置项单独建立每个虚拟机。第一组属性定义了硬件、网络和操作系统：
　　Step04使用bash和Puppet混合脚本配置应用程序配置项：
　　Vagrant文件简单定义了虚拟机管理程序级别的配置项，其余配置项则通过Puppet配置，并定义了两个层次。第一层是安装Ubuntu，为应用程序配置做准备。第二层则包含了实际的应用程序配置操作。要创建第一层的配置项，需要创建用于JDK配置的bash脚本和初始化配置项的Puppet脚本。
　　Step05在项目的scripts目录中新建installJdk.sh文件并添加以下代码：
　　我们将会使用声明式的Puppet脚本调用该脚本文件。
　　Step06在manifest目录中新建jdk.pp：
　　Step07在manifest目录中新建provisioningInit.pp文件，然后定义需要用到的包和静态变量值：
　　如果需要了解更多有关Hiera的信息，请查阅Puppet的文档页http://docs.puppetlabs.com/hiera/1/index.html。
　　Step08然后，我们就可以克隆包含二级配置项的代码仓库了：
　　Step09我们还需要配置Puppet插件Hiera，它能以分层方式分离配置脚本的各项属性：
　　Step10最后，还需要在data目录中增加几个文件。创建Hiera基本配置文件hiera.yaml：
　　Step11除此之外，你还需要host文件，它在本地集群环境中充当DNS：
　　在配置完善的环境中不需要host文件，但对于我们这个“只有一个主机”的开发环境来说，host文件还是非常实用的。
　　这样就完成了整个项目的构建，虽然它能正确配置虚拟机并安装需要用到的包，但我们还是需要创建Application层的配置项，而这些配置项包含在独立的代码仓库中。
　　Step12为当前项目初始化你的Git代码仓库，并将其提交至bitbucket.org。
　　1.5.2解析
　　我们在如图1-6所示的三个独立层次上进行配置。
　　图1-6三个独立层次
　　本节内容只涵盖了配置Guest层和Hypervisor层的方法，我们会在下一节介绍Application层的配置。分层的一个重要原因是，你通常会根据不同的部署管理程序为不同层创建不同的配置项。当配置好虚拟机后，所有环境中的应用栈配置项都应该是一致的。保持配置的一致性非常重要，因为只有在这个条件下，我们才能在完成产品开发前对我们部署的产品进行成百上千次的测试，并保证它们处于可重用和版本控制的状态下。
　　开发环境中的VirtualBox是一个虚拟机管理程序，它通过Vagrant和Puppet提供其所需的配置项。Vagrant通过扩展VirtualBox镜像实现对虚拟机的配置。VirtualBox镜像属于受版本控制的工件。Vagrant文件中定义的每个Box，都需要指定以下参数：
　　Base Box
　　网络设置
　　共享目录
　　虚拟机的内存和CPU设置
　　这些基本配置项并不包含任何产品环境中你期望的基本控制策略，比如安全、访问控制、托管和监控。所以你必须在部署至产品环境中之前配置好这些基本控制策略。你可以在Puppet Forge上找到有关内容：http://forge.puppetlabs.com。
　　然后调用配置代理来完成剩余的工作：
　　执行上面这条命令行语句会安装host文件，该文件包含了详细集群信息。
　　这条语句会更新Ubuntu apt-get缓存中所有的包，然后Vagrant会安装JDK和基本配置项。最后它将调用这些应用程序配置项。
　　虚拟机镜像可能已经包含了完整的配置，所以不一定要执行以上步骤。但需要知道如何创建合适的镜像，以及如何平衡基本镜像中特定配置的数量，否则你的配置数量会过多。
　　1.6创建Storm集群——配置Storm
　　当拥有一组准备好进行应用程序配置的虚拟机后，你就可以在每个节点上安装和配置适当的软件包了。
　　1.6.1实战
　　Step01按照图1-7所示的目录结构，新建名为storm-puppet的项目。
　　图1-7storm-puppet的目录结构
　　Step02在配置好的节点上，Puppet的运行入口（起点）是site.pp。在manifests目录下新建site.pp并添加以下内容：
　　Step03接着，你需要定义Storm模块。模块应该放置于modules目录下，并且和Puppet项目根目录的结构一样，拥有自己的manifests和template目录结构。在Storm模块目录中创建必要的清单文件（modules/storm/manifests），首先创建init.pp：
　　Step04每个storm节点上安装Storm应用程序的方法都是一样的，只是配置需要通过预先定义好的模板，根据实际情况进行调整。接着，我们来创建install.pp文件，该文件会指定需要下载和安装的二进制文件：
　　这里的install清单文件是建立在Ubuntu的Debian安装包基础之上的。它基于脚本构建，并可以根据需求进行调整。二进制文件和创建脚本可以在https://bitbucket.org/qanderson/storm-deb-packaging找到。
　　将会安装以下包：
　　Storm
　　ZeroMQ：http://www.zeromq.org/
　　Java-ZeroMQ
　　Step05每个节点的配置是通过基于模板产生的配置文件生成的。在storm清单目录下创建config.pp：
　　Step06所有storm参数都通过Hiera定义。Hiera则会使用storm清单目录下params.pp定义的Hiera配置：
　　由于属性的数量过多，因此在此省略了配置的其他内容。请到位于https://bitbucket.org/ganderson/storm-puppet/src的Git代码仓库获取完整的清单文件。
　　Step07然后指定每个节点类。我们先定义nimbus类：
　　然后是supervisor类：
　　接着是ui类：
　　最后指定zoo类（供zookeeper节点使用）：
　　Step08创建好所有文件后，初始化Git代码仓库并提交至bitbucket.org。
　　Step09为了确保配置生效，我们还要回到vagrant-storm-cluster目录并执行以下命令：
　　Step10如果你想通过ssh访问任意一个节点，只需执行下面的命令：
　　将nimbus替换为你所需的节点名。
　　1.6.2解析
　　在使用Puppet过程中，可以应用多种不同的模式。最简单的方法是使用分布式模型，在这种模式下，每一个节点进行自主配置，这与使用Puppet Master的集中式模式完全不同。 在分布式模型中，若要更新服务器配置，只需更新配置清单，并将配置清单提交到集中式Git代码仓库即可。所有节点随后会获取并应用这些配置。可以通过cron jobs、触发器，或者使用如Jenkins、Bamboo、Go这样的持续集成工具完成这项工作。使用以下命令调用Vagrant，在开发环境中进行配置：
　　这时Puppet就会应用清单文件。Puppet是一个使用声明式语言的配置管理工具，其中每一个语言元素都定义了期望的状态以及获得该状态的方法。也就是说，当系统已经处于所需状态下时，Puppet就会跳过特定的配置步骤，也就避免了重复配置的不利影响。
　　每个节点都会克隆storm-puppet项目，并在其本地应用清单配置。每个节点只应用与自己对应的配置项，而应用哪一项配置取决于site.pp清单中定义的主机名，比如：
　　在这种情况下，nimbus节点将包含cluster1的Hiera配置，然后将会安装nimbus和ui节点。由于在node定义中可以包含任意类的组合，因此你能够方便地定义整个环境。
　　1.7获取基本的点击率统计信息
　　统计点击率Topology用于记录基本的网站使用量统计信息，具体来说它会统计以下信息：
　　访客数量
　　独立访客数量
　　特定国家的访客数量
　　特定城市的访客数量
　　特定国家中每个城市访客数量的百分比
　　该系统假定访客总量一定，而且希望服务端的用户标示与客户端的Cookies是相对应的。该Topology通过IP地址和公用IP解析服务获取地理位置信息。
　　统计点击率Topology还将使用Redis存储发送到Topology的点击事件，具体来说就是将其作为一个持久化队列；除此以外还要利用Redis记录再次访问网站的访客信息。
　　关于Redis的更多信息，请访问Redis.io。
　　1.7.1准备工作
　　在我们开始之前，先要安装Redis（2.6或更高版本）：
　　然后启动Redis服务器。
　　1.7.2实战
　　Step01新建一个名为click-topology的Java项目，然后按照创建“Hello World” Topology项目的方法建立pom.xml文件和目录结构。
　　Step02在pom.xml文件中修改对应的项目名称和引用，然后在[dependencies]标签中添加以下依赖项：
　　Step03要特别注意JUnit和JMock的scope定义，避免将它们包含在用于最终部署的JAR包中。
　　Step04在src/main/java目录下的package storm.cookbook包中创建ClickTopology主类。该类定义了Topology，并提供了在集群或者本地模式下运行Topology的机制。按照下面的方式创建这个类：
　　Step05然后实现main方法，该方法根据运行时传递进来的参数个数来决定使用哪种Topology运行模式：
　　Step06这个Topology假定由Web服务器向Redis队列推送消息。因此你必须创建一个Spout，才能将这些消息作为一个流添加到Storm集群中。在storm.coobook包中创建ClickSpout类，用于在集群启动时建立与Redis的连接：
　　Step07集群会通过nextTuple方法从Spout获取新的Tuple：
　　Step08接下来，我们需要创建Bolt，其作用是通过数据库或者远程API查询来丰富基本数据。让我们从“处理重复访客信息Bolt”开始创建，该Bolt会比较客户端ID与之前访问记录是否一致，以此判断用户是否第一次访问，然后发送处理后包含这个标记的Tuple。创建RepeatVisitBolt类，提供打开逻辑和Redis连接逻辑：
　　Step09在execute方法中，Tuple由集群中的ClickSpout类提交。Bolt会基于Tuple里的字段，从Redis查询之前的访问标记，然后发送处理后的Tuple：
　　Step10接下来就该创建“处理地理位置信息Bolt”了。该Bolt会通过远程API调用查询客户IP地址所对应的国家和城市。GeographyBolt类把实际的调用委托给一个预先添加的IP解析器，以便于提高类的可测试性。在storm.cookbook包中创建GeographyBolt类，并继承BaseRichBolt接口，然后实现execute方法：
　　Step11为了能够使用这个解析器，我们需要自己实现一个名为HttpIPResolver的解析器，并在设计的时候就将其添加到GeographyBolt类的实现当中：
　　Step12接下来，我们还需要获取地理位置统计信息。GeoStatsBolt类的原理很简单，它从GeographicBolt类接收处理完成后的Tuple，与此同时在内存中维护这个数据结构。它还可以发送更新的统计信息给订阅者。GeoStatsBolt类的设计初衷是让我们能够把不同国家的总人口数据分割到不同的Bolt中进行处理。但需要确保一个国家所有城市的数据在同一个Bolt中进行处理。Topology就是根据不同国家这一点将流分割并添加到Bolt中的。
　　Step13创建GeoStatsBolt类，并实现execute方法：
　　Step14大部分逻辑都包含在内部模型类中，这个类在内存中维护城市和国家信息：
　　Step15最后，基于RepeatVisitBolt类处理完成后的流，VisitorStatsBolt方法对访客数量和独立访客数量进行最终统计。这个Bolt在内存中维护了访客与独立访客的数量，它需要接收所有的计数信息并进行统计，我们能通过Topology中的定义看出些端倪：
　　Step16现在来实现VisitorStatsBolt类，创建这个类并定义两个整型成员变量，total和uniqueCount，然后实现execute方法：
　　1.7.3解析
　　图1-8描述了统计点击率Topology。
　　图1-8统计点击率Topology
　　Spout把Web服务器上获得的点击事件发送到Topology中，再通过随机分组转发给“处理地理位置信息Bolt”和“处理重复访客信息Bolt”。这样能保证集群内的负载均衡，对于像这样处理速度慢或高延时的处理过程来说负载均衡尤为重要。
　　在设计Topology之前，需要充分理解数据模型的交换结合性质、你的流的其他问题和结构的固有模型。
　　在搭建Topology的时候，要对Storm的并行性有充分的认识。Storm wiki上有一篇文章对这个问题作了很好的总结，网址是https://github.com/nathanmarz/storm/wiki/Understanding-the-parallelism-of-a-Storm-topology。其中有几个要点需要我们注意：
　　Topology中工作进程（Worker）的数量（TOPOLOGY_WORKERS）。
　　Topology中每个组件执行单元（线程）的数量。这个值取决于并行程度。值得注意的是，这组值只是初始值（线程数），运行过程中可通过Topology重新配置（通过UI或CLI）执行单元的数量。还可以通过Config#setMaxTaskParallelism()方法限制执行单元的数量。
　　每个执行单元的默认任务数都是1。可以在声明新组件的时候通过ComponentConfigurationDeclarer#setNumTasks()方法调整这个值。
　　这些是调整集群大小时要点。集群会尝试分发任务给工作进程，每个进程包含许多执行单元，而每个执行单元又可能在运行一个或多个任务。每个工作进程的执行单元数量其实是一个执行单元总数量和执行进程总数量之比的函数。关于这个问题，可以在刚才提到的wiki页面上找到一个很好的例子。
　　可以参考节点的数量和每个节点处理器的核数来调整集群大小。我们推荐为每个线程（执行单元）分配一个核。
　　1.8对Bolt进行单元测试
　　对于任何产品交付来说，单元测试都是其重要组成部分，我们也必须对Bolt中实现的逻辑进行单元测试。
　　1.8.1准备工作
　　单元测试中有一种很常用的技术叫做Mock，它允许你动态生成依赖对象的假实例，这样才能保证对一个特定类进行准确的单元测试。本书通过JUnit 4和JMock来演示单元测试。请花些时间阅读JMock使用方法，网址是 http://jmock.org/cookbook.html。
　　1.8.2实战
　　Step01在src/test/java目录下创建storm.cookbook包，然后创建StormTestCase类。该类是一些初始化代码的抽象类：
　　Step02创建TestRepeatVisitBolt类，并继承StormTestCase，然后使用参数化执行器注解来标记它：
　　Step03execute方法中涵盖了类的测试用例逻辑：
　　Step04接下来，我们还需要定义一些参数：
　　Step05在使用Redis进行测试之前，必须初始化一些基本配置的值：
　　保留测试完成后栈中的数据总是有用的，这样便于复查和调试逻辑。我们只要在下一次测试执行时清理栈中的数据就好了。
　　1.8.3解析
　　首先，单元测试是由一组测试数据驱动的测试。这就是说我们能在测试不同用例的时候免去许多无谓和重复的操作。在测试运行之前会将静态数据写入Redis 数据库中，保证了测试的正确运行。接着测试程序就会一行一行地执行参数化的测试数据，并验证许多不同的用例。
　　JMock负责Mock收集器及Bolt发送的Tuple。然后通过Mock的对象与对象之间的交互来定义我们预期执行的操作：
　　我们应该把Expectations对象内的代码当做描述性的语言来理解。比如方法体第一行的意思就是：希望Tuple的getStringByField方法被精确地调用一次（必须只被调用一次），而且必定会返回一个值给被测试的对象。
　　这种机制能够让我们准确地测试Bolt。
　　还有很多种不同的单元测试方法，我们通常会利用数据库来实现测试。除非万不得已，还是建议Mock所有类的依赖项，这样才能实现真正意义上的单元测试。对于“地理位置信息Bolt”的resolver抽象结构来说，我们完全可以Mock所有成员。
　　1.9实现集成测试
　　由于环境和使用者的差异，集成测试可能会代表多种不同的概念。在本书中，集成测试指的是根据事先定义的输入和输出点，通过端到端的方法测试本地集群Topology的方法。这样就能确保在部署到实际的集群环境之前对Topology进行完整的功能性验证。
　　1.9.1实战
　　Step01在src/test/java目录下的storm.cookbook包中创建IntegrationTestTopology类。我们在这里添加一个“测试Bolt”类，以便于搭建本地Topology：
　　Step02然后，定义一组成对数组作为所需的参数：
　　Step03这样我们就可以根据这些参数编写测试逻辑了：
　　1.9.2解析
　　该集成测试首先会创建本地集群，然后通过Redis将输入数据添加到集群中，这个过程与实际Web服务器的工作原理一致。然后它会把预先定义好的“测试Bolt”连接到Topology结构的末尾，这样我们就能接收所有输出的Tuple并与期望值比较。
　　一旦TestBolt的值提交到了集群，在测试过程中就再也不能访问它了。所以只能通过数据持久化手段访问这些输出数据。TestBolt会把收到的Tuple数据持久化存储在Redis上，这样测试用例就可以读取并验证这些值了。TestBolt的代码逻辑如下所示：
　　然后读取这些值，并与我们所期望的值进行比较：
　　1.10将产品部署到集群
　　在完成整个开发过程之前，我们需要对集群中的Topology进行功能性测试，然后再将其部署到下个环境中。
　　1.10.1实战
　　Step01首先需要在主机开发环境中配置Storm客户端，在用户主目录下创建.storm目录。然后在这个目录下新建storm.yaml，内容如下：
　　Step02在项目根目录下，使用以下命令打包Topology：
　　Step03这样就会在项目的目标目录下生成一个完整并打包好的JAR文件。这时就可以用storm客户端命令将JAR包部署到集群上：
　　1.10.2解析
　　Storm命令行客户端为你提供了所有控制集群功能的工具。其中一个功能就是部署打包好的Topology。如果想了解更多有关storm CLI的信息，可以阅读Storm wiki上提供的详细文档，地址是https://github.com/nathanmarz/storm/wiki/Command-line-client。