用Storm轻松实时大数据分析【翻译】

原文地址

简单易用，Storm让大数据分析变得垂手可得。

现在，公司在平常运做中常常会产生TB（terabytes）级的数据。数据来源包括从网络传感器捕获的，到Web，社交媒体，交易型业务数据，以及其余业务环境中建立的数据。考虑到数据的生成量，实时计算（real-time computation ）已成为不少组织面临的一个巨大挑战。咱们已经有效地使用了一个可扩展的实时计算系统——开源的 Storm 工具，它是有 Twitter 开发，一般被称为“实时 Hadoop（real-time Hadoop）”。然而，Storm 远远比 Hadoop 简单，由于它并不须要掌握新技术处理大数据。

本文介绍了如何使用 Storm。示例项目称之为“超速警报系统（Speeding Alert System），”分析实时数据，当车速超过一个预约义的阈值（threshold）时，触发一个 trigger，相关数据就会保存到数据库中。

什么是Storm

---

Hadoop 依靠批量处理（batch processing），而 Storm 是一个实时的（real-time），分布式的（distributed），容错的（fault-tolerant），计算系统。像 Hadoop，它能够保证可靠性处理大量的数据，但不能实时；也就是说，每一个消息都将被处理。Storm 也提供这些特性，如容错，分布式计算，这些使它适合在不一样机器上处理大规模数据。它还具备以下特性：

• 简扩展。若想扩展，你只需添加设备和改变 topology 的并行性设置。用于集群协调的 Hadoop Zookeeper 用在 Storm 使得它很是容易扩展。
• 保证每一个消息都被处理。
• Storm 集群（cluster）很容易管理。
• 容错性：一旦 topology 被提交，Storm 运行 topology，直到它被杀掉或集群被关闭。此外，若是执行期间发生错误，那么从新分配的任务由 Storm 处理。
• Storm 的 topology 能够用任何语言定义，但一般仍是用 Java。

文章接下来的部分，你首先须要安装和创建 Storm。步骤以下：

• Storm 官方站点下载 Storm.
• 解压，将 bin/ 添加到你的环境变量 PATH，保证 bin/storm 脚本可执行。

Hadoop Map/Reduce 的数据处理过程是，从HDFS中获取数据，分片后，进行分布式处理，最终输出结果。

Hadoop 与 Storm 的概念对比，以下表所示：

Hadoop	Storm
JobTracker	Nimbus
TaskTracker	Supervisor
Child	Worker
Job	Topology
Mapper/Reducer	Spout/Bolt

Twitter 列举了Storm的三大类应用：

1. 信息流处理（Stream processing）。Storm可用来实时处理新数据和更新数据库，兼具容错性和可扩展性。即Storm能够用来处理源源不断流进来的消息，处理以后将结果写入到某个存储中去。

2. 连续计算（Continuous computation）。Storm可进行连续查询并把结果即时反馈给客户端。好比把Twitter上的热门话题发送到浏览器中。

3. 分布式远程程序调用（Distributed RPC）。Storm可用来并行处理密集查询。Storm的拓扑结构是一个等待调用信息的分布函数，当它收到一条调用信息后，会对查询进行计算，并返回查询结果。举个例子Distributed RPC能够作并行搜索或者处理大集合的数据。

经过配置drpc服务器，将storm的topology发布为drpc服务。客户端程序能够调用drpc服务将数据发送到storm集群中，并接收处理结果的反馈。这种方式须要drpc服务器进行转发，其中drpc服务器底层经过thrift实现。适合的业务场景主要是实时计算。而且扩展性良好，能够增长每一个节点的工做worker数量来动态扩展。

4. 项目实施，构建Topology。

Storm组件

---

Storm 集群主要由主节点（master）和工做节点（worker node）组成，它们经过 Zookeeper 进行协调。Nimbus相似Hadoop里面的 JobTracker。Nimbus 负责在集群里面分发代码，分配计算任务给机器， 而且监控状态。

主节点（master）——Nimbus

主节点运行一个守护进程（daemon），Nimbus，它负责在集群中分布代码，分配任务（Task）并监测故障。它相似于 Hadoop 的 Job Tracker。

工做者节点（worker node）——Supervisor

工做节点一样会运行一个守护进程，Supervisor，它监听已分配的工做，并按要求运行工做进程。每一个工做节点都执行一个 topology 的子集。Nimbus 和 Supervisor 之间的协调是由 Zookeeper 或其集群来管理。

Zookeeper

Zookeeper 负责 Supervisor 和 Nimbus 之间的协调。一个实时应用程序的逻辑被封装到一个 Storm 的“topology”中。一个 topology 是由一组 spouts（数据源）和 bolts（数据操做）组成，经过 Stream Groupings 链接（协调）。下面更进一步说明这些术语。

Spout

简单来讲，一个 spout 在 topology 中从一个源中读取数据。spout 能够是可靠的，也能够是不可靠的。若是 Storm 处理失败，那么一个可靠的 spout 能够确保从新发送元组（它是一个数据项的有序列表）。一个不可靠的 spout，元组一旦发送，它不会跟踪。spout 中的主要方法是 nextTuple()。该种方法或者向 topology 发出一个新元组，或是直接返回，若是没有什么可发。

Bolt

bolt 负责全部处理处理 topology 发生的一切。 bolt 可作从过滤到链接，聚合，写文件/数据库等等任何事。bolt 从一个 spout 接收数据来处理，在复杂流转换中，它能够进一步发出元组到另外一个 bolt。bolt 中主要方法是 execute()，它接受一个元组做为输入。在 spout 和 bolt，发动元组到更多的流，能够在 declareStream() 中声明和指定流。

Stream Grouping

stream grouping 定义流在 bolt 任务之间如何被划分。Storm 提供了内置的流分组：随机分组（shuffle grouping），域组域（fields grouping），全部分组（all grouping），单一分组（one grouping），直接分组（direct grouping）和本地/随机分组（local/shuffle grouping）。自定义分组实现可使用 CustomStreamGrouping 接口。

• 随机分组（Shuffle grouping）：随机分发tuple到Bolt的任务，保证每一个任务得到相等数量的tuple。
• 字段分组（Fields grouping）：根据指定字段分割数据流，并分组。例如，根据“user-id”字段，相同“user-id”的元组老是分发到同一个任务，不一样“user-id”的元组可能分发到不一样的任务。
• 所有分组（All grouping）：tuple被复制到bolt的全部任务。这种类型须要谨慎使用。
• 全局分组（Global grouping）：所有流都分配到bolt的同一个任务。明确地说，是分配给ID最小的那个task。
• 无分组（None grouping）：你不须要关心流是如何分组。目前，无分组等效于随机分组。但最终，Storm将把无分组的Bolts放到Bolts或Spouts订阅它们的同一线程去执行（若是可能）。
• 直接分组（Direct grouping）：这是一个特别的分组类型。元组生产者决定tuple由哪一个元组处理者任务接收。

另外，还涉及其余概念。

Task

worker 中每个 spout/bolt 的线程称为一个 task。在 storm0.8 以后，task 再也不与物理线程对应，同一个 spout/bolt 的 task 可能会共享一个物理线程，该线程称为 executor。

Tuple

一次消息传递的基本单元。原本应该是一个 key-value 的 map，可是因为各个组件间传递的tuple的字段名称已经事先定义好，因此，tuple 中只要按序填入各个value 就好了，因此就是一个 value list。

Topology

storm中运行的一个实时应用程序，由于各个组件间的消息流动造成逻辑上的一个拓扑结构。一个topology是spouts和bolts组成的图， 经过stream groupings将图中的spouts和bolts链接起来，以下图：

一个 topology 会一直运行，直到你 kill 掉它，Storm自动地从新分配执行失败的任务， 而且Storm能够保证你不会有数据丢失（若是开启了高可靠性的话）。若是一些机器意外停机它上面的全部任务会被转移到其余机器上。

运行一个topology很简单。首先，把你全部的代码以及所依赖的jar打进一个jar包。而后运行相似下面的这个命令：

storm jar all-my-code.jar backtype.storm.MyTopology arg1 arg2 

这个命令会运行主类: backtype.strom.MyTopology, 参数是arg1, arg2。这个类的main函数定义这个topology而且把它提交给Nimbus。storm jar负责链接到Nimbus而且上传jar包。

Topology的定义是一个Thrift结构，而且Nimbus就是一个Thrift服务， 你能够提交由任何语言建立的topology。上面的方面是用JVM-based语言提交的最简单的方法。

Stream

源源不断传递的tuple就组成了stream。消息流stream是storm里的关键抽象。一个消息流是一个没有边界的tuple序列， 而这些tuple序列会以一种分布式的方式并行地建立和处理。经过对stream中tuple序列中每一个字段命名来定义stream。在默认的状况下，tuple的字段类型能够是：integer，long，short， byte，string，double，float，boolean和byte array。你也能够自定义类型（只要实现相应的序列化器）。

每一个消息流在定义的时候会被分配给一个id，由于单向消息流使用的至关广泛， OutputFieldsDeclarer 定义了一些方法让你能够定义一个stream而不用指定这个id。在这种状况下这个stream会分配个值为‘default’默认的id 。

Storm提供的最基本的处理stream的原语是spout和bolt。你能够实现spout和bolt提供的接口来处理你的业务逻辑。

 

实现

---

对于咱们的示例中，咱们设计了一个 spout 和 bolt 的 topology，能够处理大量规模数据（日志文件），设计触发一个报警，当一个特定值超过预设阈值时。使用 Storm 的 topology，日志文件按行读取，topology 监控到来的数据。在 Storm 组件，spout 读取到来的数据。它不只从现存的文件中读取数据，也监控新文件。一旦文件被修改，spout 读取新条目，转换为元组（一个能够被 bolt 读取的格式）后，把元组发出到 bolt 执行阈值分析，查找任何超过阈值的记录。

阈值分析（Threshold Analysis）

---

本节主要集中两种类型的阈值（threshold）分析：瞬时阈值（instant thershold）和时间序列阈值（time series threshold）。

• 瞬时阈值监测：一个字段的值在那个瞬间超过了预设的临界值，若是条件符合的话则触发一个trigger。举个例子当车辆超越80千米每小时，则触发trigger。
• 时间序列阈值监测：字段的值在一个给定的时间段内超过了预设的临界值，若是条件符合则触发一个触发器。好比：在5分钟内，时速超过80千米每小时两次及以上的车辆。

清单 1 显示一个咱们使用的日志文件，它包含车辆数据信息，例如车辆号码，速度，位置。

清单 1：日志文件，经过检查点的车辆信息

AB 123, 60, North city

BC 123, 70, South city

CD 234, 40, South city

DE 123, 40, East city

EF 123, 90, South city

GH 123, 50, West city

建立相应的XML文件，它由到来的数据格式组成。用于解析日志文件。架构 XML 及其相应的描述以下表所示。

XML文件和日志文件都被 spout 随时监测，本例使用的 topology 以下图所示。

图 1：Storm中建立的 topology，以处理实时数据

如图1所示，FilelistenerSpout 接收输入日志，并逐行读取，把数据发送给 ThresoldCalculatorBolt 进一步的阈值处理。一旦处理完成，根据阈值计算的行被发动到 DBWriterBolt，持久化到数据库（或发出报警）。这个过程的具体实现将在下面介绍。

Spout 实现

---

spout 把日志文件和XML描述符文件做为输入。该XML文件指定了日志文件的格式。如今考虑一个例子的日志文件，它包含车辆信息，如车辆号码，速度，位置等三个信息。如图 2 所示。

图 2：数据从日志文件到 spout 的流程图

列表 2 显示了tuple对应的XML，其中指定了字段、将日志文件切割成字段的定界符以及字段的类型。XML文件以及数据都被保存到Spout指定的路径。

列表 2：用以描述日志文件的XML文件

<TUPLEINFO>

              <FIELDLIST>

                  <FIELD>

                            <COLUMNNAME>vehicle_number</COLUMNNAME> 

                            <COLUMNTYPE>string</COLUMNTYPE> 

                  </FIELD>

                  <FIELD>

                            <COLUMNNAME>speed</COLUMNNAME> 

                            <COLUMNTYPE>int</COLUMNTYPE> 

                  </FIELD>

                  <FIELD>

                             <COLUMNNAME>location</COLUMNNAME> 

                             <COLUMNTYPE>string</COLUMNTYPE> 

                  </FIELD>

              </FIELDLIST>  

           <DELIMITER>,</DELIMITER> 

</TUPLEINFO>

构造函数用参数 Directory、PathSpout 和 TupleInfo 对象建立 Spout 对象。TupleInfo 储存与日志文件相关的必要信息，如字段、分隔符、字段类型等。该对象经过XSTream序列化XML来建建。

Spout实现步骤：

• 监听一个单独日志文件的变化。监控目录是否添加新的日志文件。
• 声明字段后，把 spout 读取行转换成 tuple。
• 声明Spout和Bolt之间的分组，并决定tuple发送给Bolt的方式。

Spout 代码以下列表 3 所示。

列表 3：Spout中 open、nextTuple 和 delcareOutputFields 方法

public void open( Map conf, TopologyContext context,SpoutOutputCollector collector ) 

{

    _collector = collector;

    try

    {

    fileReader  =  new BufferedReader(new FileReader(new File(file)));

    } 

    catch (FileNotFoundException e) 

    {

    System.exit(1);

    }

}

public void nextTuple() 

{

    protected void ListenFile(File file) 

    {

    Utils.sleep(2000);

    RandomAccessFile access = null; 

    String line = null;                  

       try

       { 

           while ((line = access.readLine()) != null)

           { 

               if (line !=null)

               {

                    String[] fields=null;

                    if (tupleInfo.getDelimiter().equals("|"))

                       fields = line.split("\\"+tupleInfo.getDelimiter());

                    else                                                                                                             fields = line.split(tupleInfo.getDelimiter());                                                

                    if (tupleInfo.getFieldList().size() == fields.length)

                       _collector.emit(new Values(fields)); 

               }          

           } 

      } 

      catch (IOException ex) { }              

      }

}

public void declareOutputFields(OutputFieldsDeclarer declarer) 

{

     String[] fieldsArr = new String [tupleInfo.getFieldList().size()];

     for(int i=0; i<tupleInfo.getFieldList().size(); i++)

     {

         fieldsArr[i] = tupleInfo.getFieldList().get(i).getColumnName();

     }           

     declarer.declare(new Fields(fieldsArr));

}   

declareOutputFileds() 决定tuple发送的格式，这样，Bolt就能用相似的方式编码 tuple。Spout持续监听添加到日志文件的数据，一旦有数据添加，它就读取并把数据发送给 bolt 处理。

Bolt 实现

---

Spout 输出结果将给予Bolt进行更深一步的处理。通过对用例的思考，咱们的topology中须要如图 3中的两个Bolt。

图 3：Spout到Bolt的数据流程

ThresholdCalculatorBolt

Spout将tuple发出，由ThresholdCalculatorBolt接收并进行临界值处理。在这里，它将接收好几项输入进行检查；分别是：

临界值检查

• 阈值栏数检查（拆分红字段的数目）
• 阈值数据类型（拆分后字段的类型）
• 阈值出现的频数
• 阈值时间段检查

列表 4中的类，定义用来保存这些值。

public class ThresholdInfo implements Serializable

{

    private String action;

    private String rule;

    private Object thresholdValue;

    private int thresholdColNumber;

    private Integer timeWindow;

    private int frequencyOfOccurence;

}

基于字段中提供的值，阈值检查将被在 execute() 方法执行，如列表 5 所示。代码大部分的功能是解析和检测到来的值。

列表 5：阈值检测代码段

public void execute(Tuple tuple, BasicOutputCollector collector) 

{

    if(tuple!=null)

    {

        List<Object> inputTupleList = (List<Object>) tuple.getValues();

        int thresholdColNum = thresholdInfo.getThresholdColNumber();

        Object thresholdValue = thresholdInfo.getThresholdValue();

        String thresholdDataType = 

            tupleInfo.getFieldList().get(thresholdColNum-1).getColumnType();

        Integer timeWindow = thresholdInfo.getTimeWindow();

        int frequency = thresholdInfo.getFrequencyOfOccurence();

        if(thresholdDataType.equalsIgnoreCase("string"))

        {

            String valueToCheck = inputTupleList.get(thresholdColNum-1).toString();

            String frequencyChkOp = thresholdInfo.getAction();

            if(timeWindow!=null)

            {

                long curTime = System.currentTimeMillis();

                long diffInMinutes = (curTime-startTime)/(1000);

                if(diffInMinutes>=timeWindow)

                {

                    if(frequencyChkOp.equals("=="))

                    {

                         if(valueToCheck.equalsIgnoreCase(thresholdValue.toString()))

                         {

                             count.incrementAndGet();

                             if(count.get() > frequency)

                                 splitAndEmit(inputTupleList,collector);

                         }

                    }

                    else if(frequencyChkOp.equals("!="))

                    {

                        if(!valueToCheck.equalsIgnoreCase(thresholdValue.toString()))

                        {

                             count.incrementAndGet();

                             if(count.get() > frequency)

                                 splitAndEmit(inputTupleList,collector);

                         }

                     }

                     else

                         System.out.println("Operator not supported");

                 }

             }

             else

             {

                 if(frequencyChkOp.equals("=="))

                 {

                     if(valueToCheck.equalsIgnoreCase(thresholdValue.toString()))

                     {

                         count.incrementAndGet();

                         if(count.get() > frequency)

                             splitAndEmit(inputTupleList,collector);    

                     }

                 }

                 else if(frequencyChkOp.equals("!="))

                 {

                      if(!valueToCheck.equalsIgnoreCase(thresholdValue.toString()))

                      {

                          count.incrementAndGet();

                          if(count.get() > frequency)

                              splitAndEmit(inputTupleList,collector);   

                      }

                  }

              }

           }

           else if(thresholdDataType.equalsIgnoreCase("int") || 

                   thresholdDataType.equalsIgnoreCase("double") || 

                   thresholdDataType.equalsIgnoreCase("float") || 

                   thresholdDataType.equalsIgnoreCase("long") || 

                   thresholdDataType.equalsIgnoreCase("short"))

           {

               String frequencyChkOp = thresholdInfo.getAction();

               if(timeWindow!=null)

               {

                    long valueToCheck = 

                        Long.parseLong(inputTupleList.get(thresholdColNum-1).toString());

                    long curTime = System.currentTimeMillis();

                    long diffInMinutes = (curTime-startTime)/(1000);

                    System.out.println("Difference in minutes="+diffInMinutes);

                    if(diffInMinutes>=timeWindow)

                    {

                         if(frequencyChkOp.equals("<"))

                         {

                             if(valueToCheck < Double.parseDouble(thresholdValue.toString()))

                             {

                                  count.incrementAndGet();

                                  if(count.get() > frequency)

                                      splitAndEmit(inputTupleList,collector);

                             }

                         }

                         else if(frequencyChkOp.equals(">"))

                         {

                              if(valueToCheck > Double.parseDouble(thresholdValue.toString())) 

                              {

                                  count.incrementAndGet();

                                  if(count.get() > frequency)

                                      splitAndEmit(inputTupleList,collector);

                              }

                          }

                          else if(frequencyChkOp.equals("=="))

                          {

                             if(valueToCheck == Double.parseDouble(thresholdValue.toString()))

                             {

                                 count.incrementAndGet();

                                 if(count.get() > frequency)

                                     splitAndEmit(inputTupleList,collector);

                              }

                          }

                          else if(frequencyChkOp.equals("!="))

                          {

   . . . 

                          }

                      }

             }

     else

          splitAndEmit(null,collector);

     }

     else

     {

          System.err.println("Emitting null in bolt");

          splitAndEmit(null,collector);

     }

}

根据阈值 bolt 发送的 tuple 被发送到下一个相应的Bolt，在咱们的用例中是 DBWriterBolt。

DBWriterBolt

已经处理的tuple必须被持久化，以便于触发tigger或者未来使用。DBWiterBolt 完成的工做是将 tuple 持久化到数据库。表的创建是由 prepare() 完成，这也是topology调用的第一个方法。该方法的代码如列表 6 所示。

列表 6：建立表的代码

public void prepare( Map StormConf, TopologyContext context ) 

{       

    try

    {

        Class.forName(dbClass);

    } 

    catch (ClassNotFoundException e) 

    {

        System.out.println("Driver not found");

        e.printStackTrace();

    }

    try

    {

       connection driverManager.getConnection( 

           "jdbc:mysql://"+databaseIP+":"+databasePort+"/"+databaseName, userName, pwd);

       connection.prepareStatement("DROP TABLE IF EXISTS "+tableName).execute();

       StringBuilder createQuery = new StringBuilder(

           "CREATE TABLE IF NOT EXISTS "+tableName+"(");

       for(Field fields : tupleInfo.getFieldList())

       {

           if(fields.getColumnType().equalsIgnoreCase("String"))

               createQuery.append(fields.getColumnName()+" VARCHAR(500),");

           else

               createQuery.append(fields.getColumnName()+" "+fields.getColumnType()+",");

       }

       createQuery.append("thresholdTimeStamp timestamp)");

       connection.prepareStatement(createQuery.toString()).execute();

       // Insert Query

       StringBuilder insertQuery = new StringBuilder("INSERT INTO "+tableName+"(");

       String tempCreateQuery = new String();

       for(Field fields : tupleInfo.getFieldList())

       {

            insertQuery.append(fields.getColumnName()+",");

       }

       insertQuery.append("thresholdTimeStamp").append(") values (");

       for(Field fields : tupleInfo.getFieldList())

       {

           insertQuery.append("?,");

       }

       insertQuery.append("?)");

       prepStatement = connection.prepareStatement(insertQuery.toString());

    }

    catch (SQLException e) 

    {       

        e.printStackTrace();

    }       

}

数据的插入是分批次完成的。插入的逻辑由 execute() 方法提供，如列表 7 所示。大部分代码是解析各类不一样输入类型。

列表 7：数据插入的代码部分

public void execute(Tuple tuple, BasicOutputCollector collector) 

{

    batchExecuted=false;

    if(tuple!=null)

    {

       List<Object> inputTupleList = (List<Object>) tuple.getValues();

       int dbIndex=0;

       for(int i=0;i<tupleInfo.getFieldList().size();i++)

       {

           Field field = tupleInfo.getFieldList().get(i);

           try {

               dbIndex = i+1;

               if(field.getColumnType().equalsIgnoreCase("String"))             

                   prepStatement.setString(dbIndex, inputTupleList.get(i).toString());

               else if(field.getColumnType().equalsIgnoreCase("int"))

                   prepStatement.setInt(dbIndex,

                       Integer.parseInt(inputTupleList.get(i).toString()));

               else if(field.getColumnType().equalsIgnoreCase("long"))

                   prepStatement.setLong(dbIndex, 

                       Long.parseLong(inputTupleList.get(i).toString()));

               else if(field.getColumnType().equalsIgnoreCase("float"))

                   prepStatement.setFloat(dbIndex, 

                       Float.parseFloat(inputTupleList.get(i).toString()));

               else if(field.getColumnType().equalsIgnoreCase("double"))

                   prepStatement.setDouble(dbIndex, 

                       Double.parseDouble(inputTupleList.get(i).toString()));

               else if(field.getColumnType().equalsIgnoreCase("short"))

                   prepStatement.setShort(dbIndex, 

                       Short.parseShort(inputTupleList.get(i).toString()));

               else if(field.getColumnType().equalsIgnoreCase("boolean"))

                   prepStatement.setBoolean(dbIndex, 

                       Boolean.parseBoolean(inputTupleList.get(i).toString()));

               else if(field.getColumnType().equalsIgnoreCase("byte"))

                   prepStatement.setByte(dbIndex, 

                       Byte.parseByte(inputTupleList.get(i).toString()));

               else if(field.getColumnType().equalsIgnoreCase("Date"))

               {

                  Date dateToAdd=null;

                  if (!(inputTupleList.get(i) instanceof Date))  

                  {  

                       DateFormat df = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");

                       try

                       {

                           dateToAdd = df.parse(inputTupleList.get(i).toString());

                       }

                       catch (ParseException e) 

                       {

                           System.err.println("Data type not valid");

                       }

                   }  

                   else

                   {

            dateToAdd = (Date)inputTupleList.get(i);

            java.sql.Date sqlDate = new java.sql.Date(dateToAdd.getTime());

            prepStatement.setDate(dbIndex, sqlDate);

            }   

            } 

        catch (SQLException e) 

        {

             e.printStackTrace();

        }

    }

    Date now = new Date();          

    try

    {

        prepStatement.setTimestamp(dbIndex+1, new java.sql.Timestamp(now.getTime()));

        prepStatement.addBatch();

        counter.incrementAndGet();

        if (counter.get()== batchSize) 

        executeBatch();

    } 

    catch (SQLException e1) 

    {

        e1.printStackTrace();

    }           

   }

   else

   {

        long curTime = System.currentTimeMillis();

       long diffInSeconds = (curTime-startTime)/(60*1000);

       if(counter.get()<batchSize && diffInSeconds>batchTimeWindowInSeconds)

       {

            try {

                executeBatch();

                startTime = System.currentTimeMillis();

            }

            catch (SQLException e) {

                 e.printStackTrace();

            }

       }

   }

}

public void executeBatch() throws SQLException

{

    batchExecuted=true;

    prepStatement.executeBatch();

    counter = new AtomicInteger(0);

}

一旦Spout和Bolt准备就绪（等待被执行），topology生成器将会创建topology并执行。下面就来看一下执行步骤。

在本地集群上运行和测试topology

• 经过TopologyBuilder创建topology。
• 使用Storm Submitter，将topology递交给集群。以topology的名字、配置和topology的对象做为参数。
• 提交topology。

列表 8：创建和执行topology

public class StormMain

{

     public static void main(String[] args) throws AlreadyAliveException, 

                                                   InvalidTopologyException, 

                                                   InterruptedException 

     {

          ParallelFileSpout parallelFileSpout = new ParallelFileSpout();

          ThresholdBolt thresholdBolt = new ThresholdBolt();

          DBWriterBolt dbWriterBolt = new DBWriterBolt();

          TopologyBuilder builder = new TopologyBuilder();

          builder.setSpout("spout", parallelFileSpout, 1);

          builder.setBolt("thresholdBolt", thresholdBolt,1).shuffleGrouping("spout");

          builder.setBolt("dbWriterBolt",dbWriterBolt,1).shuffleGrouping("thresholdBolt");

          if(this.argsMain!=null && this.argsMain.length > 0) 

          {

              conf.setNumWorkers(1);

              StormSubmitter.submitTopology( 

                   this.argsMain[0], conf, builder.createTopology());

          }

          else

          {    

              Config conf = new Config();

              conf.setDebug(true);

              conf.setMaxTaskParallelism(3);

              LocalCluster cluster = new LocalCluster();

              cluster.submitTopology(

              "Threshold_Test", conf, builder.createTopology());

          }

     }

}

建立 topology 后，提交到本地集群。一旦topology被提交，除非被 kill 或者由于修改而关闭集群，不然它将一直运行。这也是Storm一大优点之一。

本例展现创建和使用Storm，一旦你理解 topology、spout和bolt这些基本概念，将会很容易。若是你处理大数据，又不想用 Hadoop，那么使用 Storm 是一个很好的选择。

Storm常见问题解答

---

• 1、我有一个数据文件，或者我有一个系统里面有数据，怎么导入storm作计算？

你须要实现一个Spout，Spout负责将数据emit到storm系统里，交给bolts计算。怎么实现spout能够参考官方的kestrel spout实现：

https://github.com/nathanmarz/storm-kestrel

若是你的数据源不支持事务性消费，那么就没法获得storm提供的可靠处理的保证，也不必实现ISpout接口中的ack和fail方法。

• 2、Storm为了保证tuple的可靠处理，须要保存tuple信息，这会不会致使内存OOM？

Storm为了保证tuple的可靠处理，acker会保存该节点建立的tuple id的xor值，这称为ack value，那么每ack一次，就将tuple id和ack value作异或(xor)。当全部产生的tuple都被ack的时候， ack value必定为0。这是个很简单的策略，对于每个tuple也只要占用约20个字节的内存。对于100万tuple，也才20M左右。关于可靠处理看这个：

https://github.com/nathanmarz/storm/wiki/Guaranteeing-message-processing

• 3、Storm计算后的结果保存在哪里？能够保存在外部存储吗？

Storm不处理计算结果的保存，这是应用代码须要负责的事情，若是数据不大，你能够简单地保存在内存里，也能够每次都更新数据库，也能够采用NoSQL存储。storm并无像s4那样提供一个Persist API，根据时间或者容量来作存储输出。这部分事情彻底交给用户。

数据存储以后的展示，也是你须要本身处理的，storm UI只提供对topology的监控和统计。

• 4、Storm怎么处理重复的tuple？

由于Storm要保证tuple的可靠处理，当tuple处理失败或者超时的时候，spout会fail并从新发送该tuple，那么就会有tuple重复计算的问题。这个问题是很难解决的，storm也没有提供机制帮助你解决。一些可行的策略：

（1）不处理，这也算是种策略。由于实时计算一般并不要求很高的精确度，后续的批处理计算会更正实时计算的偏差。

（2）使用第三方集中存储来过滤，好比利用mysql,memcached或者redis根据逻辑主键来去重。

（3）使用bloom filter作过滤，简单高效。

• 5、Storm的动态增删节点

我在storm和s4里比较里谈到的动态增删节点，是指storm能够动态地添加和减小supervisor节点。对于减小节点来讲，被移除的supervisor上的worker会被nimbus从新负载均衡到其余supervisor节点上。在storm 0.6.1之前的版本，增长supervisor节点不会影响现有的topology，也就是现有的topology不会从新负载均衡到新的节点上，在扩展集群的时候很不方便，须要从新提交topology。所以我在storm的邮件列表里提了这个问题，storm的开发者nathanmarz建立了一个issue 54并在0.6.1提供了rebalance命令来让正在运行的topology从新负载均衡，具体见：

https://github.com/nathanmarz/storm/issues/54

和0.6.1的变动：

http://groups.google.com/group/storm-user/browse_thread/thread/24a8fce0b2e53246

storm并不提供机制来动态调整worker和task数目。

• 6、Storm UI里spout统计的complete latency的具体含义是什么？为何emit的数目会是acked的两倍？

这个事实上是storm邮件列表里的一个问题。Storm做者marz的解答：

The complete latency is the time  from the spout emitting a tuple to that tuple being acked on the spout. So it tracks the time  for the whole tuple tree to be processed.

If you dive into the spout component in the UI, you'll see that a lot of the emitted/transferred is on the __ack* stream.  This is the spout communicating with the ackers which take care of tracking the tuple trees.

简单地说，complete latency表示了tuple从emit到被acked通过的时间，能够认为是tuple以及该tuple的后续子孙（造成一棵树）整个处理时间。其次spout的emit和transfered还统计了spout和acker之间内部的通讯信息，好比对于可靠处理的spout来讲，会在emit的时候同时发送一个_ack_init给acker，记录tuple id到task id的映射，以便ack的时候能找到正确的acker task。

其余开源的大数据解决方案

---

自 Google 在 2004 年推出 MapReduce 范式以来，已诞生了多个使用原始 MapReduce 范式（或拥有该范式的质量）的解决方案。Google 对 MapReduce 的最初应用是创建万维网的索引。尽管此应用程序仍然很流行，但这个简单模型解决的问题也正在增多。

表 1 提供了一个可用开源大数据解决方案的列表，包括传统的批处理和流式处理应用程序。在将 Storm 引入开源以前将近一年的时间里，Yahoo! 的 S4 分布式流计算平台已向 Apache 开源。S4 于 2010 年 10 月发布，它提供了一个高性能计算 (HPC) 平台，向应用程序开发人员隐藏了并行处理的复杂性。S4 实现了一个可扩展的、分散化的集群架构，并归入了部分容错功能。

表 1. 开源大数据解决方案

解决方案	开发商	类型	描述
Storm	Twitter	流式处理	Twitter 的新流式大数据分析解决方案
S4	Yahoo!	流式处理	来自 Yahoo! 的分布式流计算平台
Hadoop	Apache	批处理	MapReduce 范式的第一个开源实现
Spark	UC Berkeley AMPLab	批处理	支持内存中数据集和恢复能力的最新分析平台
Disco	Nokia	批处理	Nokia 的分布式 MapReduce 框架
HPCC	LexisNexis	批处理	HPC 大数据集群

参考资料

---

• Apache Storm 的历史及经验教训——Nathan Marz【翻译】
• Apache Storm 与 Spark：对实时处理数据，如何选择【翻译】