Storm入门教程第五章一致性事务

Storm是一个分布式的流处理系统，利用anchor和ack机制保证所有tuple都被成功处理。如果tuple出错，则可以被重传，但是如何 保证出错的tuple只被处理一次呢？Storm提供了一套事务性组件Transaction Topology，用来解决这个问题。

峨边彝族网站建设公司成都创新互联,峨边彝族网站设计制作，有大型网站制作公司丰富经验。已为峨边彝族上千家提供企业网站建设服务。企业网站搭建\外贸网站制作要多少钱，请找那个售后服务好的峨边彝族做网站的公司定做！

 Transactional Topology目前已经不再维护，由Trident来实现事务性topology，但是原理相同。

5.1一致性事务的设计

Storm如何实现即对tuple并行处理，又保证事务性。本节从简单的事务性实现方法入手，逐步引出Transactional Topology的原理。

5.1.1 简单设计一：强顺序流

保证tuple只被处理一次，最简单的方法就是将tuple流变成强顺序的，并且每次只处理一个tuple。从1开始，给每个tuple都顺序加上 一个id。在处理tuple的时候，将处理成功的tuple id和计算结果存在数据库中。下一个tuple到来的时候，将其id与数据库中的id做比较。如果相同，则说明这个tuple已经被成功处理过了，忽略 它；如果不同，根据强顺序性，说明这个tuple没有被处理过，将它的id及计算结果更新到数据库中。

以统计消息总数为例。每来一个tuple，如果数据库中存储的id 与当前tuple id不同，则数据库中的消息总数加1，同时更新数据库中的当前tuple id值。如图：

         但是这种机制使得系统一次只能处理一个tuple，无法实现分布式计算。

5.1.2 简单设计二：强顺序batch流

为了实现分布式，我们可以每次处理一批tuple，称为一个batch。一个batch中的tuple可以被并行处理。

我们要保证一个batch只被处理一次，机制和上一节类似。只不过数据库中存储的是batch id。batch的中间计算结果先存在局部变量中，当一个batch中的所有tuple都被处理完之后，判断batch id，如果跟数据库中的id不同，则将中间计算结果更新到数据库中。

如何确保一个batch里面的所有tuple都被处理完了呢？可以利用Storm提供的CoordinateBolt。如图：


但是强顺序batch流也有局限，每次只能处理一个batch，batch之间无法并行。要想实现真正的分布式事务处理，可以使用storm提供的Transactional Topology。在此之前，我们先详细介绍一下CoordinateBolt的原理。

5.1.3 CoordinateBolt原理

CoordinateBolt具体原理如下：

  • 真正执行计算的bolt外面封装了一个CoordinateBolt。真正执行任务的bolt我们称为real bolt。
  • 每个CoordinateBolt记录两个值：有哪些task给我发送了tuple（根据topology的grouping信息）；我要给哪些tuple发送信息（同样根据groping信息）
  •  Real bolt发出一个tuple后，其外层的CoordinateBolt会记录下这个tuple发送给哪个task了。
  • 等所有的tuple都发送完了之后，CoordinateBolt通过另外一个特殊的stream以emitDirect的方式告诉所有它发送过 tuple的task，它发送了多少tuple给这个task。下游task会将这个数字和自己已经接收到的tuple数量做对比，如果相等，则说明处理 完了所有的tuple。
  • 下游CoordinateBolt会重复上面的步骤，通知其下游。

整个过程如图所示：

CoordinateBolt主要用于两个场景：

  • DRPC
  • Transactional Topology

CoordinatedBolt对于业务是有侵入的，要使用CoordinatedBolt提供的功能，你必须要保证你的每个bolt发送的每个 tuple的***个field是request-id。 所谓的“我已经处理完我的上游”的意思是说当前这个bolt对于当前这个request-id所需要做的工作做完了。这个request-id在DRPC 里面代表一个DRPC请求；在Transactional Topology里面代表一个batch。

#p#

5.1.4 Trasactional Topology

Storm提供的Transactional Topology将batch计算分为process和commit两个阶段。Process阶段可以同时处理多个batch，不用保证顺序 性；commit阶段保证batch的强顺序性，并且一次只能处理一个batch，第1个batch成功提交之前，第2个batch不能被提交。

还是以统计消息总数为例，以下代码来自storm-starter里面的TransactionalGlobalCount。

MemoryTransactionalSpout spout = new MemoryTransactionalSpout(DATA,new Fields(“word“), PARTITION_TAKE_PER_BATCH);

TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder(“global-count“, “spout“, spout, 3);

builder.setBolt(“partial-count“, new BatchCount(), 5).noneGrouping(“spout“);

builder.setBolt(“sum“, new UpdateGlobalCount()).globalGrouping(“partial-count“);

TransactionalTopologyBuilder共接收四个参数。

  • 这个Transactional Topology的id。Id用来在Zookeeper中保存当前topology的进度，如果这个topology重启，可以继续之前的进度执行。
  •  Spout在这个topology中的id
  • 一个TransactionalSpout。一个Trasactional Topology中只能有一个TrasactionalSpout.在本例中是一个MemoryTransactionalSpout，从一个内存变量（DATA）中读取数据。
  • TransactionalSpout的并行度（可选）。

下面是BatchCount的定义：

 
 
    
  
  
  1. public static class BatchCount extends BaseBatchBolt { 
    2. 
  
  
  2.  
    3. 
  
  
  3.         Object _id; 
    4. 
  
  
  4.  
    5. 
  
  
  5.         BatchOutputCollector _collector; 
    6. 
  
  
  6.  
    7. 
  
  
  7.         int _count = 0; 
    8. 
  
  
  8.  
    9. 
  
  
  9.         @Override 
    10. 
  
  
  10.  
    11. 
  
  
  11.         public void prepare(Map conf, TopologyContext context, 
    12. 
  
  
  12.  
    13. 
  
  
  13.               BatchOutputCollector collector, Object id) { 
    14. 
  
  
  14.  
    15. 
  
  
  15.             _collector = collector; 
    16. 
  
  
  16.  
    17. 
  
  
  17.             _id = id; 
    18. 
  
  
  18.  
    19. 
  
  
  19.         } 
    20. 
  
  
  20.  
    21. 
  
  
  21.         @Override 
    22. 
  
  
  22.  
    23. 
  
  
  23.         public void execute(Tuple tuple) { 
    24. 
  
  
  24.  
    25. 
  
  
  25.             _count++; 
    26. 
  
  
  26.  
    27. 
  
  
  27.         } 
    28. 
  
  
  28.  
    29. 
  
  
  29.         @Override 
    30. 
  
  
  30.  
    31. 
  
  
  31.         public void finishBatch() { 
    32. 
  
  
  32.  
    33. 
  
  
  33.             _collector.emit(new Values(_id, _count)); 
    34. 
  
  
  34.  
    35. 
  
  
  35.         } 
    36. 
  
  
  36.  
    37. 
  
  
  37.         @Override 
    38. 
  
  
  38.  
    39. 
  
  
  39.         public void declareOutputFields(OutputFieldsDeclarer declarer) { 
    40. 
  
  
  40.  
    41. 
  
  
  41.             declarer.declare(new Fields(“id“, “count“)); 
    42. 
  
  
  42.  
    43. 
  
  
  43.         } 
    44. 
  
  
  44.  
    45. 
  
  
  45. } 
    46.

 BatchCount的prepare方法的***一个参数是batch id，在Transactional Tolpoloyg里面这id是一个TransactionAttempt对象。

Transactional Topology里发送的tuple都必须以TransactionAttempt作为***个field，storm根据这个field来判断tuple属于哪一个batch。

TransactionAttempt包含两个值：一个transaction id，一个attempt id。transaction id的作用就是我们上面介绍的对于每个batch中的tuple是唯一的，而且不管这个batch replay多少次都是一样的。attempt id是对于每个batch唯一的一个id， 但是对于同一个batch，它replay之后的attempt id跟replay之前就不一样了， 我们可以把attempt id理解成replay-times， storm利用这个id来区别一个batch发射的tuple的不同版本。

execute方法会为batch里面的每个tuple执行一次，你应该把这个batch里面的计算状态保持在一个本地变量里面。对于这个例子来说， 它在execute方法里面递增tuple的个数。

***， 当这个bolt接收到某个batch的所有的tuple之后， finishBatch方法会被调用。这个例子里面的BatchCount类会在这个时候发射它的局部数量到它的输出流里面去。

#p#

下面是UpdateGlobalCount类的定义：

 
 
    
  
  
  1. public static class UpdateGlobalCount extends BaseTransactionalBolt 
    2. 
  
  
  2.  
    3. 
  
  
  3. implements ICommitter { 
    4. 
  
  
  4.  
    5. 
  
  
  5.         TransactionAttempt _attempt; 
    6. 
  
  
  6.  
    7. 
  
  
  7.         BatchOutputCollector _collector; 
    8. 
  
  
  8.  
    9. 
  
  
  9.         int _sum = 0; 
    10. 
  
  
  10.  
    11. 
  
  
  11.         @Override 
    12. 
  
  
  12.  
    13. 
  
  
  13.         public void prepare(Map conf, TopologyContext context, 
    14. 
  
  
  14.  
    15. 
  
  
  15. BatchOutputCollector collector, TransactionAttempt attempt) { 
    16. 
  
  
  16.  
    17. 
  
  
  17.             _collector = collector; 
    18. 
  
  
  18.  
    19. 
  
  
  19.             _attempt = attempt; 
    20. 
  
  
  20.  
    21. 
  
  
  21.         } 
    22. 
  
  
  22.  
    23. 
  
  
  23.         @Override 
    24. 
  
  
  24.  
    25. 
  
  
  25.         public void execute(Tuple tuple) { 
    26. 
  
  
  26.  
    27. 
  
  
  27.             _sum+=tuple.getInteger(1); 
    28. 
  
  
  28.  
    29. 
  
  
  29.         } 
    30. 
  
  
  30.  
    31. 
  
  
  31.         @Override 
    32. 
  
  
  32.  
    33. 
  
  
  33.         public void finishBatch() { 
    34. 
  
  
  34.  
    35. 
  
  
  35.             Value val = DATABASE.get(GLOBAL_COUNT_KEY); 
    36. 
  
  
  36.  
    37. 
  
  
  37.             Value newval; 
    38. 
  
  
  38.  
    39. 
  
  
  39.             if(val == null || !val.txid.equals(_attempt.getTransactionId())) { 
    40. 
  
  
  40.  
    41. 
  
  
  41.                 newval = new Value(); 
    42. 
  
  
  42.  
    43. 
  
  
  43.                 newval.txid = _attempt.getTransactionId(); 
    44. 
  
  
  44.  
    45. 
  
  
  45.                 if(val==null) { 
    46. 
  
  
  46.  
    47. 
  
  
  47.                     newval.count = _sum; 
    48. 
  
  
  48.  
    49. 
  
  
  49.                 } else { 
    50. 
  
  
  50.  
    51. 
  
  
  51.                     newval.count = _sum + val.count; 
    52. 
  
  
  52.  
    53. 
  
  
  53.                 } 
    54. 
  
  
  54.  
    55. 
  
  
  55.                 DATABASE.put(GLOBAL_COUNT_KEY, newval); 
    56. 
  
  
  56.  
    57. 
  
  
  57.             } else { 
    58. 
  
  
  58.  
    59. 
  
  
  59.                 newval = val; 
    60. 
  
  
  60.  
    61. 
  
  
  61.             } 
    62. 
  
  
  62.  
    63. 
  
  
  63.             _collector.emit(new Values(_attempt, newval.count)); 
    64. 
  
  
  64.  
    65. 
  
  
  65.         } 
    66. 
  
  
  66.  
    67. 
  
  
  67.         @Override 
    68. 
  
  
  68.  
    69. 
  
  
  69.         public void declareOutputFields(OutputFieldsDeclarer declarer) { 
    70. 
  
  
  70.  
    71. 
  
  
  71.             declarer.declare(new Fields(“id“, “sum“)); 
    72. 
  
  
  72.  
    73. 
  
  
  73.         } 
    74. 
  
  
  74.  
    75. 
  
  
  75. } 
    76.

UpdateGlobalCount实现了ICommitter接口，所以storm只会在commit阶段执行finishBatch方法。而execute方法可以在任何阶段完成。

在UpdateGlobalCount的finishBatch方法中，将当前的transaction id与数据库中存储的id做比较。如果相同，则忽略这个batch；如果不同，则把这个batch的计算结果加到总结果中，并更新数据库。

Transactional Topolgy运行示意图如下：

下面总结一下Transactional Topology的一些特性

  •  Transactional Topology将事务性机制都封装好了，其内部使用CoordinateBolt来保证一个batch中的tuple被处理完。
  •  TransactionalSpout只能有一个，它将所有tuple分为一个一个的batch，而且保证同一个batch的transaction id始终一样。
  •  BatchBolt处理batch在一起的tuples。对于每一个tuple调用execute方法，而在整个batch处理完成的时候调用finishBatch方法。
  •  如果BatchBolt被标记成Committer，则只能在commit阶段调用finishBolt方法。一个batch的commit阶 段由storm保证只在前一个batch成功提交之后才会执行。并且它会重试直到topology里面的所有bolt在commit完成提交。
  •  Transactional Topology隐藏了anchor/ack框架，它提供一个不同的机制来fail一个batch，从而使得这个batch被replay。

5.2 Trident介绍

Trident是Storm之上的高级抽象，提供了joins，grouping，aggregations，fuctions和filters等接口。如果你使用过Pig或Cascading，对这些接口就不会陌生。

Trident将stream中的tuples分成batches进行处理，API封装了对这些batches的处理过程，保证tuple只被处理一次。处理batches中间结果存储在TridentState对象中。

Trident事务性原理这里不详细介绍，有兴趣的读者请自行查阅资料。

参考：http://xumingming.sinaapp.com/736/twitter-storm-transactional-topolgoy/

http://xumingming.sinaapp.com/811/twitter-storm-code-analysis-coordinated-bolt/

https://github.com/nathanmarz/storm/wiki/Trident-tutorial

网页名称：Storm入门教程第五章一致性事务
路径分享：http://www.shufengxianlan.com/qtweb/news37/402887.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

网站内链知识

同城分类信息

免备案虚拟主机　　 　成都网站设计　　 　眉山主机托管　　 　阿坝网站设计　　 　网站制作　　 　蜀锦在线　　 　网站定制　　 　深圳惠宜防蚊纱窗　　 　成都小程序开发　　 　成都app开发　　 　成都装饰装修　　 　成都网站推广　　 　成都标识标牌设计制作　　 　雪茄烟品牌　　 　外贸网站建设　　 　成都橡塑保温板　　 　网络推广公司　　 　响应式网站建设　　 　香港云主机　　 　成都集团网站建设