系统基本架构



整个实时分析系统的架构就是先由电商系统的订单服务器产生订单日志, 然后使用Flume去监听订单日志，并实时把每一条日志信息抓取下来并存进Kafka消息系统中, 接着由Storm系统消费Kafka中的消息，同时消费记录由Zookeeper集群管理，这样即使Kafka宕机重启后也能找到上次的消费记录，接着从上次宕机点继续从Kafka的Broker中进行消费。但是由于存在先消费后记录日志或者先记录后消费的非原子操作，如果出现刚好消费完一条消息并还没将信息记录到Zookeeper的时候就宕机的类似问题，或多或少都会存在少量数据丢失或重复消费的问题, 其中一个解决方案就是Kafka的Broker和Zookeeper都部署在同一台机子上。接下来就是使用用户定义好的Storm Topology去进行日志信息的分析并输出到Redis缓存数据库中(也可以进行持久化)，最后用Web APP去读取Redis中分析后的订单信息并展示给用户。之所以在Flume和Storm中间加入一层Kafka消息系统，就是因为在高并发的条件下, 订单日志的数据会井喷式增长，如果Storm的消费速度(Storm的实时计算能力那是最快之一,但是也有例外, 而且据说现在Twitter的开源实时计算框架Heron比Storm还要快)慢于日志的产生速度，加上Flume自身的局限性，必然会导致大量数据滞后并丢失，所以加了Kafka消息系统作为数据缓冲区，而且Kafka是基于log File的消息系统，也就是说消息能够持久化在硬盘中，再加上其充分利用Linux的I/O特性,提供了可观的吞吐量。架构中使用Redis作为数据库也是因为在实时的环境下，Redis具有很高的读写速度。

业务背景
各大电商网站在合适的时间进行各种促销活动已是常态，在能为网站带来大量的流量和订单的同时，对于用户也有不小的让利，必然是大家伙儿喜闻乐见的。在促销活动期间，老板和运营希望能实时看到订单情况，老板开心，运营也能根据实时的订单数据调整运营策略，而让用户能实时看到网站的订单数据，也会勾起用户的购买欲。但是普通的离线计算系统已然不能满足在高并发环境下的实时计算要求，所以我们得使用专门实时计算系统,如:Storm, Heron, Spark Stream等,去满足类似的需求。
既然要分析订单数据，那必然在订单产生的时候要把订单信息记录在日志文件中。本文中，作者通过使用log4j2，以及结合自己之前开发电商系统的经验，写了一个订单日志生成模拟器，代码如下，能帮助大家随机产生订单日志。下面所展示的订单日志文件格式和数据就是我们本文中的分析目标，本文的案例中用来分析所有商家的订单总销售额并找出销售额钱20名的商家。

订单数据格式：
orderNumber: XX | orderDate: XX | paymentNumber: XX | paymentDate: XX | merchantName: XX | sku: [ skuName: XX skuNum: XX skuCode: XX skuPrice: XX totalSkuPrice: XX;skuName: XX skuNum: XX skuCode: XX skuPrice: XX totalSkuPrice: XX;] | price: [ totalPrice: XX discount: XX paymentPrice: XX ]



订单日志生成程序：
使用log4j2将日志信息写入文件中，每小时滚动一次日志文件
[plain] view plain copy
  1. <?xml version="1.0" encoding="UTF-8"?>  
  2. <Configuration status="INFO">  
  3.       <Appenders>  
  4.         <Console name="myConsole" target="SYSTEM_OUT">  
  5.           <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} [%t] %-5level %logger{36} - %msg%n"/>  
  6.         </Console>  
  7.         <RollingFile name="myFile" fileName="/Users/guludada/Desktop/logs/app.log"  
  8.           filePattern="/Users/guludada/Desktop/logs/app-%d{yyyy-MM-dd-HH}.log">  
  9.             <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} [%t] %-5level %logger{36} - %msg%n"/>  
  10.             <Policies>  
  11.                 <TimeBasedTriggeringPolicy />  
  12.             </Policies>  
  13.         </RollingFile>                  
  14.       </Appenders>  
  15.       <Loggers>  
  16.         <Root level="Info">  
  17.           <AppenderRef ref="myConsole"/>  
  18.           <AppenderRef ref="myFile"/>  
  19.         </Root>  
  20.       </Loggers>  
  21. </Configuration>  
生成器代码:
[plain] view plain copy
  1. package com.guludada.ordersInfo;  
  2.   
  3. import java.text.SimpleDateFormat;  
  4. import java.util.Date;  
  5. import java.util.Random;  
  6.   
  7. // Import log4j classes.  
  8. import org.apache.logging.log4j.LogManager;  
  9. import org.apache.logging.log4j.Logger;  
  10.   
  11.   
  12.   
  13. public class ordersInfoGenerator {  
  14.       
  15.     public enum paymentWays {  
  16.         Wechat,Alipay,Paypal  
  17.     }  
  18.     public enum merchantNames {  
  19.         优衣库,天猫,淘宝,咕噜大大,快乐宝贝,守望先峰,哈毒妇,Storm,Oracle,Java,CSDN,跑男,路易斯威登,  
  20.         暴雪公司,Apple,Sumsam,Nissan,Benz,BMW,Maserati  
  21.     }  
  22.       
  23.     public enum productNames {  
  24.         黑色连衣裙, 灰色连衣裙, 棕色衬衫, 性感牛仔裤, 圆脚牛仔裤,塑身牛仔裤, 朋克卫衣,高腰阔腿休闲裤,人字拖鞋,  
  25.         沙滩拖鞋  
  26.     }  
  27.       
  28.     float[] skuPriceGroup = {299,399,699,899,1000,2000};  
  29.     float[] discountGroup = {10,20,50,100};  
  30.     float totalPrice = 0;  
  31.     float discount = 0;  
  32.     float paymentPrice = 0;  
  33.       
  34.     private static final Logger logger = LogManager.getLogger(ordersInfoGenerator.class);  
  35.     private int logsNumber = 1000;  
  36.       
  37.     public void generate() {  
  38.                   
  39.         for(int i = 0; i <= logsNumber; i++) {             
  40.             logger.info(randomOrderInfo());           
  41.         }  
  42.     }  
  43.       
  44.     public String randomOrderInfo() {  
  45.           
  46.         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");       
  47.         Date date = new Date();       
  48.           
  49.         String orderNumber = randomNumbers(5) + date.getTime();  
  50.           
  51.         String orderDate = sdf.format(date);  
  52.           
  53.         String paymentNumber = randomPaymentWays() + "-" + randomNumbers(8);  
  54.           
  55.         String paymentDate = sdf.format(date);  
  56.           
  57.         String merchantName = randomMerchantNames();  
  58.           
  59.         String skuInfo = randomSkus();  
  60.           
  61.         String priceInfo = calculateOrderPrice();  
  62.           
  63.         return "orderNumber: " + orderNumber + " | orderDate: " + orderDate + " | paymentNumber: " +  
  64.             paymentNumber + " | paymentDate: " + paymentDate + " | merchantName: " + merchantName +   
  65.             " | sku: " + skuInfo + " | price: " + priceInfo;  
  66.     }  
  67.           
  68.     private String randomPaymentWays() {  
  69.           
  70.         paymentWays[] paymentWayGroup = paymentWays.values();  
  71.         Random random = new Random();  
  72.         return paymentWayGroup[random.nextInt(paymentWayGroup.length)].name();  
  73.     }  
  74.       
  75.     private String randomMerchantNames() {  
  76.           
  77.         merchantNames[] merchantNameGroup = merchantNames.values();  
  78.         Random random = new Random();  
  79.         return merchantNameGroup[random.nextInt(merchantNameGroup.length)].name();  
  80.     }  
  81.       
  82.     private String randomProductNames() {  
  83.           
  84.         productNames[] productNameGroup = productNames.values();  
  85.         Random random = new Random();  
  86.         return productNameGroup[random.nextInt(productNameGroup.length)].name();  
  87.     }  
  88.       
  89.       
  90.     private String randomSkus() {  
  91.           
  92.         Random random = new Random();  
  93.         int skuCategoryNum = random.nextInt(3);  
  94.           
  95.         String skuInfo ="[";  
  96.           
  97.         totalPrice = 0;  
  98.         for(int i = 1; i <= 3; i++) {  
  99.               
  100.             int skuNum = random.nextInt(3)+1;  
  101.             float skuPrice = skuPriceGroup[random.nextInt(skuPriceGroup.length)];  
  102.             float totalSkuPrice = skuPrice * skuNum;              
  103.             String skuName = randomProductNames();  
  104.             String skuCode = randomCharactersAndNumbers(10);  
  105.             skuInfo += " skuName: " + skuName + " skuNum: " + skuNum + " skuCode: " + skuCode  
  106.                     + " skuPrice: " + skuPrice + " totalSkuPrice: " + totalSkuPrice + ";";        
  107.             totalPrice += totalSkuPrice;  
  108.         }  
  109.           
  110.           
  111.         skuInfo += " ]";  
  112.           
  113.         return skuInfo;  
  114.     }  
  115.       
  116.     private String calculateOrderPrice() {  
  117.           
  118.         Random random = new Random();  
  119.         discount = discountGroup[random.nextInt(discountGroup.length)];  
  120.         paymentPrice = totalPrice - discount;  
  121.           
  122.         String priceInfo = "[ totalPrice: " + totalPrice + " discount: " + discount + " paymentPrice: " + paymentPrice +" ]";  
  123.           
  124.         return priceInfo;  
  125.     }  
  126.       
  127.     private String randomCharactersAndNumbers(int length) {  
  128.           
  129.         String characters = "abcdefghijklmnopqrstuvwxyz0123456789";  
  130.         String randomCharacters = "";    
  131.                 Random random = new Random();    
  132.                 for (int i = 0; i < length; i++) {    
  133.               randomCharacters += characters.charAt(random.nextInt(characters.length()));    
  134.                 }    
  135.                 return randomCharacters;    
  136.     }  
  137.       
  138.     private String randomNumbers(int length) {  
  139.           
  140.         String characters = "0123456789";  
  141.         String randomNumbers = "";     
  142.                 Random random = new Random();    
  143.                 for (int i = 0; i < length; i++) {    
  144.              randomNumbers += characters.charAt(random.nextInt(characters.length()));    
  145.                 }    
  146.                return randomNumbers;          
  147.     }  
  148.       
  149.     public static void main(String[] args) {  
  150.           
  151.         ordersInfoGenerator generator = new ordersInfoGenerator();  
  152.         generator.generate();  
  153.     }  
  154. }  

收集日志数据
采集数据的方式有多种，一种是通过自己编写shell脚本或Java编程采集数据，但是工作量大，不方便维护，另一种就是直接使用第三方框架去进行日志的采集，一般第三方框架的健壮性，容错性和易用性都做得很好也易于维护。本文采用第三方框架Flume进行日志采集，Flume是一个分布式的高效的日志采集系统，它能把分布在不同服务器上的海量日志文件数据统一收集到一个集中的存储资源中，Flume是Apache的一个顶级项目，与Kafka也有很好的兼容性。不过需要注意的是Flume并不是一个高可用的框架，这方面的优化得用户自己去维护。

Flume的agent是运行在JVM上的，所以各个服务器上的JVM环境必不可少。每一个Flume agent部署在一台服务器上，Flume会收集web server产生的日志数据，并封装成一个个的事件发送给Flume Agent的Source，Flume Agent Source会消费这些收集来的数据事件（Flume Event）并放在Flume Agent Channel，Flume Agent Sink会从Channel中收集这些采集过来的数据，要么存储在本地的文件系统中要么作为一个消费资源分给下一个装在分布式系统中其它服务器上的Flume Agent进行处理。Flume提供了点对点的高可用的保障，某个服务器上的Flume Agent Channel中的数据只有确保传输到了另一个服务器上的Flume Agent Channel里或者正确保存到了本地的文件存储系统中，才会被移除。

 在本文中，Flume的Source我们选择的是Exec Source，因为是实时系统，直接通过tail 命令来监听日志文件，而在Kafka的Broker集群端的Flume我们选择Kafka Sink 来把数据下沉到Kafka消息系统中。

下图是来自Flume官网里的Flume拉取数据的架构图：

    图片来源：http://flume.apache.org/FlumeUserGuide.html

订单日志产生端的Flume配置文件如下：
[plain] view plain copy
  1. agent.sources = origin  
  2. agent.channels = memorychannel  
  3. agent.sinks = target  
  4.   
  5. agent.sources.origin.type = exec  
  6. agent.sources.origin.command = tail -F /export/data/trivial/app.log  
  7. agent.sources.origin.channels = memorychannel  
  8.   
  9. agent.sources.origin.interceptors = i1  
  10. agent.sources.origin.interceptors.i1.type = static  
  11. agent.sources.origin.interceptors.i1.key = topic  
  12. agent.sources.origin.interceptors.i1.value = ordersInfo  
  13.   
  14. agent.sinks.loggerSink.type = logger  
  15. agent.sinks.loggerSink.channel = memorychannel  
  16.   
  17. agent.channels.memorychannel.type = memory  
  18. agent.channels.memorychannel.capacity = 10000  
  19.   
  20. agent.sinks.target.type = avro  
  21. agent.sinks.target.channel = memorychannel  
  22. agent.sinks.target.hostname = 172.16.124.130  
  23. agent.sinks.target.port = 4545  

Kafka 消息系统端Flume配置文件
[plain] view plain copy
  1. agent.sources = origin  
  2. agent.channels = memorychannel  
  3. agent.sinks = target  
  4.   
  5. agent.sources.origin.type = avro  
  6. agent.sources.origin.channels = memorychannel  
  7. agent.sources.origin.bind = 0.0.0.0  
  8. agent.sources.origin.port = 4545  
  9.   
  10. agent.sinks.loggerSink.type = logger  
  11. agent.sinks.loggerSink.channel = memorychannel  
  12.   
  13. agent.channels.memorychannel.type = memory  
  14. agent.channels.memorychannel.capacity = 5000000  
  15. agent.channels.memorychannel.transactionCapacity = 1000000  
  16.   
  17. agent.sinks.target.type = org.apache.flume.sink.kafka.KafkaSink  
  18. #agent.sinks.target.topic = bigdata  
  19. agent.sinks.target.brokerList=localhost:9092  
  20. agent.sinks.target.requiredAcks=1  
  21. agent.sinks.target.batchSize=100  
  22. agent.sinks.target.channel = memorychannel  

这里需要注意的是,在日志服务器端的Flume agent中我们配置了一个interceptors，这个是用来为Flume Event(Flume Event就是拉取到的一行行的日志信息)的头部添加key为“topic”的K-V键值对，这样这条抓取到的日志信息就会根据topic的值去到Kafka中指定的topic消息池中，当然还可以为Flume Event额外配置一个key为“Key”的键值对，Kafka Sink会根据key“Key”的值将这条日志信息下沉到不同的Kafka分片上，否则就是随机分配。在Kafka集群端的Flume配置里，有几个重要的参数需要注意，“topic”是指定抓取到的日志信息下沉到Kafka哪一个topic池中，如果之前Flume发送端为Flume Event添加了带有topic的头信息，则这里可以不用配置；brokerList就是配置Kafka集群的主机地址和端口；requireAcks=1是配置当下沉到Kafka的消息储存到特定partition的leader中成功后就返回确认消息，requireAcks=0是不需要确认消息成功写入Kafka中，requireAcks=-1是指不光需要确认消息被写入partition的leander中，还要确认完成该条消息的所有备份；batchSize配置每次下沉多少条消息，每次下沉的数量越多延迟也高。

Kafka消息系统
这一部分我们将谈谈Kafka的配置和使用，Kafka在我们的系统中实际上就相当于起到一个数据缓冲池的作用, 有点类似于ActiveQ的消息队列和Redis这样的缓存区的作用，但是更可靠，因为是基于log File的消息系统，数据不容易丢失，以及能记录数据的消费位置并且用户还可以自定义消息消费的起始位置，这就使得重复消费消息也可以得以实现，而且同时具有队列和发布订阅两种消息消费模式，十分灵活，并且与Storm的契合度很高，充分利用Linux系统的I/O提高读写速度等等。另一个要提的方面就是Kafka的Consumer是pull-based模型的，而Flume是push-based模型。push-based模型是尽可能大的消费数据，但是当生产者速度大于消费者时数据会被覆盖。而pull-based模型可以缓解这个压力，消费速度可以慢于生产速度，有空余时再拉取那些没拉取到的数据。

Kafka是一个分布式的高吞吐量的消息系统，同时兼有点对点和发布订阅两种消息消费模式。Kafka主要由Producer，Consumer和Broker组成。Kafka中引入了一个叫“topic”的概念，用来管理不同种类的消息，不同类别的消息会记录在到其对应的topic池中，而这些进入到topic中的消息会被Kafka写入磁盘的log文件中进行持久化处理。Kafka会把消息写入磁盘的log file中进行持久化对于每一个topic 里的消息log文件，Kafka都会对其进行分片处理，而每一个 消息都会顺序写入中log分片中，并且被标上“offset”的标量来代表这条消息在这个分片中的顺序，并且这些写入的消息无论是内容还是顺序都是不可变的。所以Kafka和其它消息队列系统的一个区别就是它能做到分片中的消息是能顺序被消费的，但是要做到全局有序还是有局限性的，除非整个topic只有一个log分片。并且无论消息是否有被消费，这条消息会一直保存在log文件中，当留存时间足够长到配置文件中指定的retention的时间后，这条消息才会被删除以释放空间。对于每一个Kafka的Consumer，它们唯一要存的Kafka相关的元数据就是这个“offset”值，记录着Consumer在分片上消费 到了哪一个位置。通常Kafka是使用Zookeeper来为每一个Consumer保存它们的offset信息，所以在启动Kafka之前需要有一个Zookeeper集群;而且Kafka默认采用的是先记录offset再读取数据的策略，这种策略会存在少量数据丢失的可能。不过用户可以灵活设置Consumer的“offset”的位置，在加上消息记录在log文件中，所以是可以重复消费消息的。log的分片和它们的备份会分散保存在集群的服务器上，对于每一个partition，在集群上都会有一台这个partition存在的服务器作为leader，而这个partitionpartition的其它备份所在的服务器做为follower，leader负责处理关于这个partition的所有请求，而follow er负责这个partition的其它备份的同步工作，当leader服务器宕机时，其中一个follower服务器就会被选举为新的leader。

一般的消息系统分为两种模式，一种是点对点的消费模式，也就是queuing模式，另一种是发布订阅模式，也就是publish-subscribe模式，而Kafka引入了一个Consumer Group的概念，使得其能兼有两种模式。在Kafka中，每一个consumer都会标明自己属于哪个consumer group，每个topic的消息都会分发给每一个subscribe了这个topic的所有consumer group中的一个consumer实例。所以当所有的consumers都在同一个consumer group中，那么就像queuing的消息系统，一个message一次只被一个consumer消费。如果每一个consumer都有不同consumer group，那么就像public-subscribe消息系统一样，一个消息分发给所有的consumer实例。对于普通的消息队列系统，可能存在多个consumer去同时消费message，虽然message是有序地分发出去的，但是由于网络延迟的时候到达不同的consumer的时间不是顺序的，这时就失去了顺序性，解决方案是只用一个consumer去消费message，但显然不太合适。而对于Kafka来说，一个partiton只分发给每一个consumer group中的一个consumer实例，也就是说这个partition只有一个consumer实例在消费，所以可以保证在一个partition内部数据的处理是有序的，不同之处就在于Kafka内部对消息进行了分片处理，虽然看上去也是单consumer的做法，但是分片机制保证了并发消费。如果要做到全局有序，那么整个topic中的消息只有一个分片，并且每一个consumer group中只能有一个consumer实例。这实际上就是彻底牺牲了消息消费时的并发度。
Kafka的配置和部署十分简单
1. 首先启动Zookeeper集群，Kafka需要Zookeeper集群来帮助记录每一个Consumer的offset
2. 为集群上的每一台Kafka服务器单独配置配置文件，比如我们需要设置有两个节点的Kafka集群，那么节点1和节点2的最基本的配置如下：
[plain] view plain copy
  1. config/server-1.properties:  
  2.     broker.id=1  
  3.     listeners=PLAINTEXT://:9093  
  4.     log.dir=export/data/kafka  
  5.     zookeeper.connect=localhost:2181  
[plain] view plain copy
  1. config/server-2.properties:  
  2.     broker.id=2  
  3.     listeners=PLAINTEXT://:9093  
  4.     log.dir=/export/data/kafka  
  5.     zookeeper.connect=localhost:2181  
b roker.id是kafka集群上每一个节点的单独标识，不能重复;listeners可以理解为每一个节点上 Kafka进程要监听的端口 ，使用默认的就行; log.dir是Kafka的log文件(记录消息的log file)存放目录; zookeeper.connect就是Zookeeper的URI地址和端口。
3. 配置完上面的配置文件后，只要分别在节点上输入下面命令启动Kafka进程就可以使用了
 
> bin/kafka-server-start.sh config/server-1.properties &
...
> bin/kafka-server-start.sh config/server-2.properties &
...

Storm实时计算框架
接下来开始介绍本篇文章要使用的实时计算框架 Storm 。 Strom 是一个非常快的实时计算框架，至于快到什么程度呢？官网首页给出的数据是每一个 Storm 集群上的节点每一秒能处理一百万条数据。相比 Hadoop 的 “Mapreduce” 计算框架， Storm 使用的是 "Topology" ； Mapreduce 程序在计算完成后最终会停下来，而 Topology 则是会永远运行下去除非你显式地使用 “kill -9 XXX” 命令停掉它。和大多数的集群系统一样， Storm 集群也存在着 Master 节点和 Worker 节点，在 Master 节点上运行的一个守护进程叫 “Nimbus” ，类似于 Hadoop 的 “JobTracker” 的功能，负责集群中计算程序的分发，任务的分发，监控任务和工作节点的运行情况等； Worker 节点上运行的守护进程叫 “Supervisor” ，负责接收 Nimbus 分发的任务并运行，每一个 Worker 上都会运行着 Topology 程序的一部分，而一个 Topology 程序的运行就是由集群上多个 Worker 一起协同工作的。值得注意的是 Nimubs 和 Supervisor 之间的协调工作也是通过 Zookeeper 来管理的， Nimbus 和 Supervisor 自己本身在集群上都是无状态的，它们的状态都保存在 Zookeeper 上，所以任何节点的宕机和动态扩容都不会影响整个集群的工作运行，并支持 fast-fail 机制。

Storm 有一个很重要的对数据的抽象概念，叫做 “Stream” ，我们姑且称之为数据流，数据流 Stream 就是由之间没有任何关系的松散的一个一个的数据元组 “tuples” 所组成的序列。要在 Storm 上做实时计算，首先你得有一个计算程序，这就是 “Topology” ，一个 Topology 程序由 “Spout” 和 “Bolt” 共同组成。 Storm 就是通过 Topology 程序将数据流 Stream 通过可靠 (ACK 机制 ) 的分布式计算生成我们的目标数据流 Stream ，就比如说把婚恋网站上当日注册的所有用户信息数据流 Stream 通过 Topology 程序计算出月收入上万年龄在 30 岁以下的新的用户信息流 Stream 。在我们的文章中， Spout 就是实现了特定接口的 Java 类，它相当于数据源，用于产生数据或者从外部接收数据；而 Bolt 就是实现了 Storm Bolt 接口的 Java 类，用于消费从 Spout 发送出来的数据流并实现用户自定义的数据处理逻辑；对于复杂的数据处理，可以定义多个连续的 Bolt 去协同处理。最后在程序中通过 Spout 和 Bolt 生成 Topology 对象并提交到 Storm 集群上执行。

tuples是Storm的数据模型,，由值和其所对应的field所组成，比如说在Spout或Bolt中定义了发出的元组的field为：(name,age,gender)，那么从这个Spout或Bolt中发出的数据流的每一个元组值就类似于(''咕噜大大",27,"中性")。 在 Storm 中还有一个 Stream Group 的概念，它用来决定从 Spout 或或或 Bolt 组件中发出的 tuples 接下来应该传到哪一个组件中或者更准确地说在程序里设置某个组件应该接收来自哪一个组件的 tuples; 并且在 Storm 中提供了多个用于数据流分组的机制，比如说 shuffleGrouping ，用来将当前组件产生的 tuples 随机分发到下一个组件中，或者 fieldsGrouping ，根据 tuples 的 field 值来决定当前组件产生的 tuples 应该分发到哪一个组件中。

另一部分需要了解的就是 Storm 中 tasks 和 workers 的概念。每一个 worker 都是一个运行在物理机器上的 JVM 进程，每个 worker 中又运行着多个 task 线程，这些 task 线程可能是 Spout 任务也可能是 Bolt 任务，由 Nimbus 根据 RoundRobin 负载均衡策略来分配，而至于在整个 Topology 程序里要起几个 Spout 线程或 Bolt 线程，也就是 tasks ，由用户在程序中设置并发度来决定。


Storm集群的配置文件如下：
Storm的配置文件在项目的conf目录下，也就是:conf/storm.yaml
[plain] view plain copy
  1. # Licensed to the Apache Software Foundation (ASF) under one  
  2. # or more contributor license agreements.  See the NOTICE file  
  3. # distributed with this work for additional information  
  4. # regarding copyright ownership.  The ASF licenses this file  
  5. # to you under the Apache License, Version 2.0 (the  
  6. # "License"); you may not use this file except in compliance  
  7. # with the License.  You may obtain a copy of the License at  
  8. #  
  9. # http://www.apache.org/licenses/LICENSE-2.0  
  10. #  
  11. # Unless required by applicable law or agreed to in writing, software  
  12. # distributed under the License is distributed on an "AS IS" BASIS,  
  13. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
  14. # See the License for the specific language governing permissions and  
  15. # limitations under the License.  
  16.   
  17. ########### These MUST be filled in for a storm configuration  
  18. storm.zookeeper.servers:  
  19.   - "ymhHadoop"  
  20.   - "ymhHadoop2"  
  21.   - "ymhHadoop3"      
  22.   
  23. storm.local.dir: "/export/data/storm/workdir"  
  24.    
  25. nimbus.host: "ymhHadoop"  
  26.   
  27. supervisor.slots.ports:  
  28.   -6700  
  29.   -6701  
  30.   -6702  
  31.   -6703   
  32.    
storm.zookeeper.servers 自然就是用来配置我们熟悉的 Zookeeper 集群中各个节点的 URI 地址和端口的
storm.local.dir 是用来配置 storm 节点相关文件的存储目录的，每一个 storm 集群的节点在本地服务器上都要有一个目录存储少量的和该节点有关的一些信息。记得要开发这个目录的读写权限哦
nimbus.host 自然就是用来指定 nimbus 服务器的 URI 的
supervisor.slots.ports 这个是用来配置 supervisor 服务器启动的 worker 所监听的端口，每一个 worker 就是一个物理的 JVM 进程。

上面这些是基本配置，并且要严格按照上面的格式来，少一个空格都会报错。
接下来就是将配置文件拷贝到集群的各个机器上，然后在分别在 nimbus 和 supervisor 机器上通过 $bin/storm nimbus 和 $bin/storm supervisor 命令来启动集群上的机子。最后在 nimbus 上通过 $bin/storm UI 命令可以启动 Storm 提供的 UI 界面，功能十分强大，可以监控集群上各个节点的运行状态，提交 Topology 任务，监控 Topology 任务的运行情况等。 这个 UI 界面可以通过 http://{nimbus host}:8080 的地址访问到。


Redis数据库
Redis 是一个基于内存的多种数据结构的存储工具，经常有人说 Redis 是一个基于 key-value 数据结构的缓存数据库，这种说法必然是不准确的， Key-Value 只是其中的一种数据结构的实现， Redis 支持 Strings ， hashes ， lists ， sets ， sorted sets 等多种常见的数据结构，并提供了功能强大的范围查询，以及提供了 INCR ， INCRBY,DECR,DECRBY 等多种原子命令操作，保证在并发的环境下不会出现脏数据。虽然 Redis 是基于内存的数据库，但也提供了多种硬盘持久化策略，比如说 RDB 策略，用来将某个时间点的 Redis 的数据快照存储在硬盘中，或者是 AOF 策略，将每一个 Redis 操作命令都不可变的顺序记录在 log 文件中，恢复数据时就将 log 文件中的所有命令顺序执行一遍等等。 Redis 不光可以作为网站热点数据的缓存服务器，还可以用来做数据库，或者消息队列服务器的 broker 等。在本文中选择 Redis 作为订单分析结果的存储工具，一方面是其灵活的数据结构和强大的数据操作命令，另一方面就是在大数据的实时计算环境下，需要 Redis 这样的具备高速 I/O 的数据库。

在本文的例子中，作者使用 Sorted Sets 数据结构来存储各个商家的总订单销售额， Sorted Sets 数据结构由 Key, Score ， element value 三部分组成， Set 的数据结构保证同一个 key 中的元素值不会重复，而在 Sorted Sets 结构中是通过 Score 来为元素值排序，这很自然地就能将各个商家的总订单销售额设置为 Score ，然后商家名称为 element value ，这样就能根据总订单销售额来为商家排序。在 Storm 程序中，我们通过 Jedis API 来调用 Redis 的
$ZINCRBY KEY INCREMENT MEMBER
的命令来统计商家总销售额 , ZINCRBY 是一个原子命令，能保证在 Storm 的并发计算的环境下，正确地增加某个商家的 Score 的值，也就是它们的订单总销售额。而对于两个商家同名这种情况应该在业务系统中去避免而不应该由我们的数据分析层来处理。最后提一个小 trips ，就是如果所有商家的 Score 都设置成相同的分数，那么 Redis 就会默认使用商家名的字母字典序来排序。


Kafka+Storm+Redis的整合
当数据被 Flume 拉取进 Kafka 消息系统中，我们就可以使用 Storm 来进行消费， Redis 来对结果进行存储。 Storm 对 Kafka 有很好的兼容性，我们可以通过 Kafka Spout 来从 Kafka 中获取数据；在 Bolt 处理完数据后，通过 Jedis API 在程序中将数据存储在 Redis 数据库中。

下面就是 Kafka Spout 和创建 Topology 的程序代码:

 BrokerHosts hosts = new ZkHosts("ymhHadoop:2181,ymhHadoop2:2181,ymhHadoop3:2181");
zkHosts 是用来指定 Zookeeper 集群的节点的 URI 和端口，而 Zookeeper 集群是用来记录 Spout 对 Kafka 消息消费的 offset 位置

spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
 主要是用来将 Spout 从 Kafka 拉取来的 byte[] 数组格式的数据转化为 Storm 的 tuples

[plain] view plain copy
  1. package com.guludada.ordersanalysis;  
  2.   
  3. import java.util.UUID;  
  4.   
  5. import backtype.storm.Config;  
  6. import backtype.storm.LocalCluster;  
  7. import backtype.storm.StormSubmitter;  
  8. import backtype.storm.generated.AlreadyAliveException;  
  9. import backtype.storm.generated.InvalidTopologyException;  
  10. import backtype.storm.spout.SchemeAsMultiScheme;  
  11. import backtype.storm.topology.TopologyBuilder;  
  12. import backtype.storm.tuple.Fields;  
  13. import storm.kafka.Broker;  
  14. import storm.kafka.BrokerHosts;  
  15. import storm.kafka.KafkaSpout;  
  16. import storm.kafka.SpoutConfig;  
  17. import storm.kafka.StaticHosts;  
  18. import storm.kafka.StringScheme;  
  19. import storm.kafka.ZkHosts;  
  20. import storm.kafka.trident.GlobalPartitionInformation;  
  21.   
  22. public class ordersAnalysisTopology {  
  23.       
  24.     private static String topicName = "ordersInfo";  
  25.     private static String zkRoot = "/stormKafka/"+topicName;  
  26.       
  27.     public static void main(String[] args) {  
  28.           
  29.         BrokerHosts hosts = new ZkHosts("ymhHadoop:2181,ymhHadoop2:2181,ymhHadoop3:2181");  
  30.   
  31.           
  32.         SpoutConfig spoutConfig = new SpoutConfig(hosts,topicName,zkRoot,UUID.randomUUID().toString());  
  33.         spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());  
  34.         KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);  
  35.           
  36.         TopologyBuilder builder = new TopologyBuilder();          
  37.         builder.setSpout("kafkaSpout",kafkaSpout);          
  38.         builder.setBolt("merchantsSalesBolt", new merchantsSalesAnalysisBolt(), 2).shuffleGrouping("kafkaSpout");  
  39.   
  40.         Config conf = new Config();  
  41.         conf.setDebug(true);  
  42.           
  43.         if(args != null && args.length > 0) {  
  44.             conf.setNumWorkers(1);  
  45.             try {  
  46.                 StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());  
  47.             } catch (AlreadyAliveException e) {  
  48.                 // TODO Auto-generated catch block  
  49.                 e.printStackTrace();  
  50.             } catch (InvalidTopologyException e) {  
  51.                 // TODO Auto-generated catch block  
  52.                 e.printStackTrace();  
  53.             }  
  54.               
  55.         } else {  
  56.               
  57.             conf.setMaxSpoutPending(3);  
  58.               
  59.             LocalCluster cluster = new LocalCluster();  
  60.             cluster.submitTopology("ordersAnalysis", conf, builder.createTopology());  
  61.               
  62.               
  63.         }  
  64.   
  65.     }  
  66. }  

下面是Bolt程序，主要是 用来处理 从Kafka拉取到的订单日志信息 , 并计算出所有商家的总订单收入，然后使用Jedis API将计算结果存入到Redis数据库中。
[plain] view plain copy
  1. package com.guludada.domain;  
  2.   
  3. import java.util.ArrayList;  
  4. import java.util.Date;  
  5.   
  6. public class ordersBean {  
  7.   
  8.     Date createTime = null;  
  9.     String number = "";  
  10.     String paymentNumber = "";  
  11.     Date paymentDate = null;  
  12.     String merchantName = "";  
  13.     ArrayList<skusBean> skuGroup = null;  
  14.     float totalPrice = 0;  
  15.     float discount = 0;  
  16.     float paymentPrice = 0;  
  17.       
  18.     public Date getCreateTime() {  
  19.         return createTime;  
  20.     }  
  21.     public void setCreateTime(Date createTime) {  
  22.         this.createTime = createTime;  
  23.     }  
  24.     public String getNumber() {  
  25.         return number;  
  26.     }  
  27.     public void setNumber(String number) {  
  28.         this.number = number;  
  29.     }  
  30.     public String getPaymentNumber() {  
  31.         return paymentNumber;  
  32.     }  
  33.     public void setPaymentNumber(String paymentNumber) {  
  34.         this.paymentNumber = paymentNumber;  
  35.     }  
  36.     public Date getPaymentDate() {  
  37.         return paymentDate;  
  38.     }  
  39.     public void setPaymentDate(Date paymentDate) {  
  40.         this.paymentDate = paymentDate;  
  41.     }  
  42.     public String getMerchantName() {  
  43.         return merchantName;  
  44.     }  
  45.     public void setMerchantName(String merchantName) {  
  46.         this.merchantName = merchantName;  
  47.     }  
  48.     public ArrayList<skusBean> getSkuGroup() {  
  49.         return skuGroup;  
  50.     }  
  51.     public void setSkuGroup(ArrayList<skusBean> skuGroup) {  
  52.         this.skuGroup = skuGroup;  
  53.     }  
  54.     public float getTotalPrice() {  
  55.         return totalPrice;  
  56.     }  
  57.     public void setTotalPrice(float totalPrice) {  
  58.         this.totalPrice = totalPrice;  
  59.     }  
  60.     public float getDiscount() {  
  61.         return discount;  
  62.     }  
  63.     public void setDiscount(float discount) {  
  64.         this.discount = discount;  
  65.     }  
  66.     public float getPaymentPrice() {  
  67.         return paymentPrice;  
  68.     }  
  69.     public void setPaymentPrice(float paymentPrice) {  
  70.         this.paymentPrice = paymentPrice;  
  71.     }  
  72.       
  73.       
  74. }  
本文 例子 中用不到sku sbean，所以这里作者就没有写委屈偷懒一下下
[plain] view plain copy
  1. package com.guludada.domain;  
  2.   
  3. public class skusBean {  
  4.       ………………  
  5. }  

logInfoHandler用来过滤订单的日志信息，并保存到ordersBean和skusBean中，方便Bolt获取日志数据的各项属性进行处理
[plain] view plain copy
  1. package com.guludada.common;  
  2.   
  3. import java.text.ParseException;  
  4. import java.text.SimpleDateFormat;  
  5. import java.util.regex.Matcher;  
  6. import java.util.regex.Pattern;  
  7.   
  8. import com.guludada.domain.ordersBean;  
  9.   
  10. public class logInfoHandler {  
  11.       
  12.     SimpleDateFormat sdf_final = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");  
  13.       
  14.     public ordersBean getOrdersBean(String orderInfo) {  
  15.           
  16.         ordersBean order = new ordersBean();  
  17.           
  18.         //从日志信息中过滤出订单信息  
  19.         Pattern orderPattern = Pattern.compile("orderNumber:.+");  
  20.         Matcher orderMatcher = orderPattern.matcher(orderInfo);  
  21.         if(orderMatcher.find()) {  
  22.               
  23.             String orderInfoStr = orderMatcher.group(0);  
  24.             String[] orderInfoGroup = orderInfoStr.trim().split("\\|");  
  25.               
  26.             //获取订单号  
  27.             String orderNum = (orderInfoGroup[0].split(":"))[1].trim();  
  28.             order.setNumber(orderNum);  
  29.                           
  30.             //获取创建时间  
  31.             String orderCreateTime = orderInfoGroup[1].trim().split(" ")[1] + " " + orderInfoGroup[1].trim().split(" ")[2];  
  32.             try {  
  33.                 order.setCreateTime(sdf_final.parse(orderCreateTime));  
  34.             } catch (ParseException e) {  
  35.                 // TODO Auto-generated catch block  
  36.                 e.printStackTrace();  
  37.             }  
  38.               
  39.             //获取商家名称  
  40.             String merchantName = (orderInfoGroup[4].split(":"))[1].trim();  
  41.             order.setMerchantName(merchantName);  
  42.               
  43.             //获取订单总额  
  44.             String orderPriceInfo = (orderInfoGroup[6].split("price:"))[1].trim();  
  45.             String totalPrice = (orderPriceInfo.substring(2, orderPriceInfo.length()-3).trim().split(" "))[1];  
  46.             order.setTotalPrice(Float.parseFloat(totalPrice));  
  47.                           
  48.             return order;  
  49.                           
  50.         } else {  
  51.             return order;  
  52.         }  
  53.     }  
  54. }  

[plain] view plain copy
  1. package com.guludada.ordersanalysis;  
  2.   
  3. import java.util.Map;  
  4.   
  5. import com.guludada.common.logInfoHandler;  
  6. import com.guludada.domain.ordersBean;  
  7.   
  8. import backtype.storm.task.OutputCollector;  
  9. import backtype.storm.task.TopologyContext;  
  10. import backtype.storm.topology.OutputFieldsDeclarer;  
  11. import backtype.storm.topology.base.BaseRichBolt;  
  12. import backtype.storm.tuple.Tuple;  
  13. import redis.clients.jedis.Jedis;  
  14. import redis.clients.jedis.JedisPool;  
  15. import redis.clients.jedis.JedisPoolConfig;  
  16.   
  17. public class merchantsSalesAnalysisBolt extends BaseRichBolt {  
  18.       
  19.     private OutputCollector _collector;  
  20.     logInfoHandler loginfohandler;  
  21.     JedisPool pool;  
  22.   
  23.     public void execute(Tuple tuple) {  
  24.         String orderInfo = tuple.getString(0);  
  25.         ordersBean order = loginfohandler.getOrdersBean(orderInfo);  
  26.           
  27.         //store the salesByMerchant infomation into Redis  
  28.         Jedis jedis = pool.getResource();  
  29.         jedis.zincrby("orderAna:topSalesByMerchant", order.getTotalPrice(), order.getMerchantName());  
  30.     }  
  31.   
  32.     public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {  
  33.         this._collector = collector;  
  34.         this.loginfohandler = new logInfoHandler();  
  35.         this.pool = new JedisPool(new JedisPoolConfig(), "ymhHadoop",6379,2 * 60000,"12345");  
  36.           
  37.     }  
  38.   
  39.     public void declareOutputFields(OutputFieldsDeclarer arg0) {  
  40.         // TODO Auto-generated method stub  
  41.           
  42.     }  
  43.   
  44. }  

Topology项目的Maven配置文件
[plain] view plain copy
  1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
  2.   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">  
  3.   <modelVersion>4.0.0</modelVersion>  
  4.   <groupId>com.guludada</groupId>  
  5.   <artifactId>Storm_OrdersAnalysis</artifactId>  
  6.   <packaging>war</packaging>  
  7.   <version>0.0.1-SNAPSHOT</version>  
  8.   <name>Storm_OrdersAnalysis Maven Webapp</name>  
  9.   <url>http://maven.apache.org</url>  
  10.   <dependencies>  
  11.     <dependency>  
  12.         <groupId>org.apache.storm</groupId>  
  13.         <artifactId>storm-core</artifactId>  
  14.         <version>0.9.6</version>  
  15.         <scope>provided</scope>  
  16.     </dependency>  
  17.     <dependency>  
  18.         <groupId>org.apache.storm</groupId>  
  19.         <artifactId>storm-kafka</artifactId>  
  20.         <version>0.9.6</version>  
  21.     </dependency>  
  22.     <dependency>  
  23.         <groupId>org.apache.kafka</groupId>  
  24.         <artifactId>kafka_2.10</artifactId>  
  25.         <version>0.9.0.1</version>  
  26.             <exclusions>  
  27.                 <exclusion>  
  28.                     <groupId>org.apache.zookeeper</groupId>  
  29.                     <artifactId>zookeeper</artifactId>  
  30.                 </exclusion>  
  31.                 <exclusion>  
  32.                     <groupId>log4j</groupId>  
  33.                     <artifactId>log4j</artifactId>  
  34.                 </exclusion>  
  35.                 <exclusion>  
  36.                     <groupId>org.slf4j</groupId>  
  37.                     <artifactId>slf4j-log4j12</artifactId>  
  38.                 </exclusion>  
  39.             </exclusions>  
  40.     </dependency>  
  41.     <dependency>  
  42.         <groupId>redis.clients</groupId>  
  43.         <artifactId>jedis</artifactId>  
  44.         <version>2.8.1</version>  
  45.     </dependency>   
  46.   </dependencies>  
  47.   <build>  
  48.     <finalName>Storm_OrdersAnalysis</finalName>  
  49.     <plugins>  
  50.         <plugin>  
  51.             <artifactId>maven-assembly-plugin</artifactId>  
  52.             <configuration>  
  53.                 <descriptorRefs>    
  54.                     <descriptorRef>jar-with-dependencies</descriptorRef>  
  55.                 </descriptorRefs>  
  56.                 <archive>  
  57.                    <manifest>  
  58.                      <mainClass>com.guludada.ordersanalysis.ordersAnalysisTopology</mainClass>  
  59.                    </manifest>  
  60.                  </archive>  
  61.              </configuration>  
  62.           </plugin>  
  63.       </plugins>  
  64.   </build>  
  65. </project>  

maven 配置文件中配置了一个官方推荐的 maven-assembly-plugin 插件，用来帮助用户方便地打包 Topology 程序的。只需要进入到项目的根路径，然后运行
$mvn assembly:assembly
 命令就可以打包好 Topology 的 jar 包了。

最后我带大家梳理一下整个项目的部署流程
1.  启动 Zookeeper
2. 启动 Kafka
3. 启动 Flume 将程序拉取到 Kafka 中
4. 启动 Storm 集群
5. 启动 Redis 服务端    通过命令
$ src/redis-server
6. 提交打包好的 Topology 程序到 Storm 集群中 通过 Storm UI 或者 命令 $storm jar path/to/allmycode.jar org.me.MyTopology arg1 arg2 arg3
7. 启动 Redis 的 CLI 客户端查看结果 通过命令
$ src/redis-cli --raw
$  zrange key 0 -1 withscores
如下图:

Troubleshooting
  1. 在使用maven同时导入storm-core, storm-kaka和kafka的依赖包的时候可能会出现jar包冲突导致无法初始化Log4jLoggerFactory，并无法启动Storm程序.解决方法也很简单，按照红字提示，把多余的jar包移除就行了，通过在maven的pom文件中kafka的依赖设置部分加入下面的设置<exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion>
  2. 第一次执行Storm建立Topology时，作者遇到了一个十分低级的问题，就是发现明明Kafka的topic里有数据，可是Storm程序怎么都无法读取到数据，后来才从下面的文章中明白了问题的所在 http://m.blog.csdn.net/article/details?id=18615761  原因就在于Topology第一次启动前还没有在zookeeper中的zkRoot创建offset信息，Storm取不到offset信息就会使用默认的offset，也就是log文件中从最后一个元素开始读取信息，所以之前在kafka中的数据都无法读出来。Storm启动后，再往broker中写数据，这些后写的数据就能正确被Storm处理。                                  
  3. 当Storm的topology传到Nimbus的时候，或者说你的Storm程序刚开始启动的时候可能会报关于JedisPool是一个无法序列化的对象而导致的错误：java.lang.RuntimeException:java.io.NotSerializableException: redis.clients.jedis.JedisPool 解决方案就是将Bolt类中外部的JedisPool初始化代码放入Bolt的prepare()方法中，如本文的代码示例所示
  4. 在Storm启动并开始连接Redis的时候，会报出连接被拒绝，因为Redis运行在protect mode模式下的错误。这是因为Storm程序是远程连接Redis的服务器端，如果Redis服务器端没有设置密码的话是拒绝远程连接的。解决方法也十分简单，关闭protect mode模式(强烈不推荐),或者使用下面命令为Redis设置密码就可以了$config set requirepass 123
  5. 向Storm提交Topology以后, Supervisor端会一直报“Kill XXXX No Such process”的错误，多数原因是提交的topology没有正确被执行，而Storm的日记中不会显示topology程序里的错误。解决方法就是启动Storm UI, 通过这个Storm自带的UI界面查看topology的运行情况，并且程序中的错误也会在UI界面中显示出来，能方便地查看topology程序的错误。

    6.kafka使用的时候的小问题：
        当在一台机子上启动kafka producer客户端的时候，是无法在同一台机子上继续启动kafka的consumer客户端的,因为这两个进程可能占用的同一个端口，需要在另外一台机子上启动kafka consumer程序，这样就能看见正确的结果了
Logo
Cloudpods

开源、云原生的融合云平台

更多推荐

面向未来的 IT 基础设施管理架构——融合云（Unified IaaS）

随着数字化时代的到来，IT系统已成为人类社会正常运转不可或缺的组成部分。不远的未来，智能制造，5G和人工智能等技术将成为推动生产力发展的重要引擎，人类社会将面临前所未有的全面彻底的数字化浪潮。IT基础设施作为IT系统运行的平台和载体，是实现数字化的基石。在这场数字化浪潮中，企业必须积极拥抱云计算技术，采用符合技术发展趋势、面向未来的IT基础构架，才能在未来的竞争中赢得先机。 一、云计算历经十余年

avatar Cloudpods

Cloudpods负载均衡的功能介绍

作者:周有松 今天的内容会从以下几个方面展开： 负载均衡产品简介。主要介绍负载均衡作为一个云上产品，它的功能模型是怎样的，日常使用中会遇到的业务词汇负载均衡的功能与典型应用场景。这部分主要结合业务词汇，对负载均衡服务中常见的一些功能选项进行介绍，并举例介绍一些典型的应用场景最后，我们做一下总结，讨论一下负载均衡产品相比传统方式的优点 一、产品简介​ 1. 以NGINX为例​ 提到负载均衡，我们以

avatar Cloudpods

使用Linux vfio将Nvidia GPU透传给QEMU虚拟机

Linux 上虚拟机 GPU 透传需要使用 vfio 的方式。主要是因为在 vfio 方式下对虚拟设备的权限和 DMA 隔离上做的更好。但是这么做也有个缺点，这个物理设备在主机和其他虚拟机都不能使用了。 qemu 直接使用物理设备本身命令行是很简单的，关键在于事先在主机上对系统、内核和物理设备的一些配置。 单纯从 qemu 的命令行来看，其实和普通虚拟机启动就差了最后那个-device的选项。这

avatar Cloudpods