Storm-HBase集成--配置和开发

港控/mmm° 2022-06-12 10:36 338阅读 0赞

1 Storm0.9.3中的对HBase的集成

Storm新版本0.9.3中重新整理和加入了对[Hbase][]的集成模块，除了基本的Bolt和Spout之外，加入了用于访问[hbase][Hbase]的Trident。利用这个Trident，我们可以更加快速的编写Storm访问HBase的代码逻辑。

关于Storm-HBase模块中，几个主要的功能类如下：

<table style="margin:0px 0px 10px; padding:0px; border:0px currentColor; color:rgb(0,0,0); line-height:13pt; clear:left; font-size:10pt; border-collapse:collapse"> 
 <tbody> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <th style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); text-align:left; line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em; background-color:rgb(240,240,240)"> 类名</th> 
   <th style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); text-align:left; line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em; background-color:rgb(240,240,240)"> 介绍</th> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.trident.mapper.TridentHBaseMapper</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> 把HBase的Row Key, 列簇和列，对应到Storm Trident中Tuple的Field</td> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.trident.mapper.SimpleTridentHBaseMapper</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <span style="margin-top:0px; margin-bottom:0px">上面的HBaseMapper的一个简单继承实现，指定Rowkey的Field和列簇/列的Field</span></td> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.bolt.mapper.HBaseValueMapper</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> 用户继承这个类来实现把HBase的Cell映射成Storm中的Tuple，这个类通常是被继承后使用</td> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.trident.state.HBaseUpdater</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> 更新HBaseState的类</td> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.trident.state.HBaseState</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> Trident中负责HBase数据的状态类</td> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.trident.state.HBaseStateFactory</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> 工厂类，负责生产HBaseState对象</td> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> 负责定义HBase数据到Storm Tuple的投影，需要指定HBase的表名，列族名和列</td> 
  </tr> 
  <tr style="line-height:13pt; font-size:10pt"> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> <pre style="padding:0px; overflow:visible; margin-top:0px; margin-bottom:0px">org.apache.storm.hbase.security.HBaseSecurityUtil</pre> </td> 
   <td style="margin:0px; padding:5px 7px; border:1px solid rgb(221,221,221); line-height:13pt; font-size:10pt; vertical-align:top; min-width:0.6em"> 专门用于让Storm通过HBase集群的Kerberos安全验证的类</td> 
  </tr> 
 </tbody> 
</table>

### 2 代码示例 ###

下面是一个完整的从Kafka中读取数据，并写入HBase的代码（Trident模式）。

**\[html\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1.  // Storm Tuple中的两个Field，分别叫做word 和 count  
2.  Fields fields = new Fields("word", "count");  
3.     
4.  // 定义HBase配置相关和Kerberos相关  
5.  String hBaseConfigKey = "config\_key";  
6.  System.setProperty("java.security.krb5.realm", "HADOOP.QIYI.COM");  
7.  System.setProperty("java.security.krb5.kdc", "kerberos-hadoop-dev001-shjj.qiyi.virtual");  
8.     
9.  //载入HBase和Kerberos相关配置，Config对象是来自backtype.storm.Config 类  
10. Config conf = new Config();  
11. conf.setDebug(true);  
12. Map<String, String> hBaseConfigMap = new HashMap<String, String>();  
13. hBaseConfigMap.put(HBaseSecurityUtil.STORM\_KEYTAB\_FILE\_KEY, "/home/yeweichen/yeweichen.keytab");  
14. hBaseConfigMap.put(HBaseSecurityUtil.STORM\_USER\_NAME\_KEY, "yeweichen@HADOOP.QIYI.COM");  
15. conf.put("config\_key", hBaseConfigMap);  
16.    
17. // 定义Trident拓扑，从Kafka中获取数据  
18. TridentTopology tridentTopology = new TridentTopology();  
19. BrokerHosts zk = new ZkHosts("10.121.43.14,10.121.43.17");  
20. TridentKafkaConfig spoutConf = new TridentKafkaConfig(zk, "mytopic");  
21. spoutConf.forceFromStart = true;  
22. spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());  
23. OpaqueTridentKafkaSpout spout = new OpaqueTridentKafkaSpout(spoutConf);  
24.    
25. //定义HBase的Mapper，指定“word”字段的内容作为rowkey，列族名为cf  
26. TridentHBaseMapper tridentHBaseMapper = new SimpleTridentHBaseMapper()  
27.         .withColumnFamily("cf")  
28.         .withColumnFields(new Fields("word"))  
29.         .withColumnFields(new Fields("count"))  
30.         .withRowKeyField("word");  
31.    
32. // LogCollect就是自定义的Mapper  
33. HBaseValueMapper rowToStormValueMapper = new LogCollectMapper();  
34.    
35. //定义投影类，加入cf列族中的word和count两个列  
36. HBaseProjectionCriteria projectionCriteria = new HBaseProjectionCriteria();  
37. projectionCriteria.addColumn(new HBaseProjectionCriteria.ColumnMetaData("cf", "word"));  
38. projectionCriteria.addColumn(new HBaseProjectionCriteria.ColumnMetaData("cf", "count"));  
39.    
40. //定义HBaseState类的属性类Option  
41. HBaseState.Options options = new HBaseState.Options()  
42.         .withConfigKey(hBaseConfigKey)  
43.         .withDurability(Durability.SYNC\_WAL)  
44.         .withMapper(tridentHBaseMapper)  
45.         .withProjectionCriteria(projectionCriteria)  
46.         .withRowToStormValueMapper(rowToStormValueMapper)  
47.         .withTableName("storminput");  
48.    
49. //使用工厂方法和Option生成HBaseState对象  
50. StateFactory factory = new HBaseStateFactory(options);  
51.    
52. //定义Stream，从Kafka中读出的数据，使用AddTimeFunction方法把它生成word和field两个字段，然后把他们写入HBase,如上面定义的，word字段作为row key  
53. tridentTopology.newStream("myspout", spout).each(new Fields("str"), new AddTimeFunction(), new Fields("word", "count"))  
54.         .partitionPersist(factory, fields, new HBaseUpdater(), new Fields());  
55.    
56. // 提交拓扑  
57. StormSubmitter.submitTopology(args\[0\], conf,tridentTopology.build());

[Hbase]: http://lib.csdn.net/base/hbase
[view plain]: http://blog.csdn.net/liuxiao723846/article/details/52351829#

发表评论取消回复

表情：

评论列表（有 0 条评论，338人围观）

还没有评论，来说两句吧...

相关阅读

相关 Android Java集成开发环境（IDE）配置详解

在配置Android Java集成开发环境（IDE，通常指Eclipse或 IntelliJ IDEA）时，主要关注以下几个方面： 1. **安装IDE**： - **

川长思鸟来/ 2024年09月13日 01:30/ 0 赞/ 80 阅读

相关安装配置Android Studio集成开发环境

目录一、下载Android Studio 二、安装Android Studio （一）进入安装向导（二）选择安装组件（三）选择安装位置（四）选择开始菜单文件

Myth丶恋晨/ 2024年04月05日 07:25/ 0 赞/ 135 阅读

相关 Eclipse+Maven+Tomcat 集成开发环境配置

在Eclipse中创建的Dynamic Web Project 类型的Web 项目，通过Run As -> Run on Server 可以添加本地安装的Tomcat，在Ec

客官°小女子只卖身不卖艺/ 2024年03月31日 15:52/ 0 赞/ 114 阅读

相关 Swagger的配置和集成

Swagger 首先我们需要了解swagger 面向所有人的API开发借助Swagger开源和专业工具集，为用户，团队和企业简化API开发号称世界

迷南。/ 2023年07月09日 15:27/ 0 赞/ 16 阅读

相关 jersey和spring集成配置使用

jersey 是基于Java的一个轻量级RESTful风格的Web Services框架。 [官网][Link 1] 使用maven，在pom.xml中加入：

朱雀/ 2022年07月14日 06:39/ 0 赞/ 201 阅读

相关 Storm-HBase集成--配置和开发

1 Storm0.9.3中的对HBase的集成 Storm新版本0.9.3中重新整理和加入了对[Hbase][]的集成模块，除了基本的Bolt和Spout之外，加入了

港控/mmm°/ 2022年06月12日 10:36/ 0 赞/ 339 阅读

相关 Facebook Android集成和开发

关键字：Facebook | Android studio 3.0 Facebook离线包如何集成？公司的环境决定了无法通过仓库的方法进行集成，因此只能下载Facebook

亦凉/ 2022年05月29日 04:28/ 0 赞/ 248 阅读

相关 RabbitMQ 安装配置和 Spring 集成

本文从安装和配置 RabbitMQ 开始，准备好环境后，直接在 Spring 中集成，并且针对 Spring 中的常见用法提供了示例和讲解。 > 由于我个人更多的在 cent

一时失言乱红尘/ 2022年04月21日 18:16/ 0 赞/ 252 阅读

相关 Spring项目集成和配置superdiamond方式

如果Java项目系统中使用到了superdiamond统一配置管理平台来对不同版本的项目包做配置信息管理，以下是Spring集成superdiamond的配置方式：在相关的

淩亂°似流年/ 2022年03月18日 02:54/ 0 赞/ 287 阅读

相关配置IDLE(python自带的集成开发环境)

设置代码自动补全在使用IDLE编写代码时，当我们引用模块函数时会出现延时弹出代码补齐或直接不出现的现象，这需要更改代码补全的配置文件。在你python的安装目录下：Pyth

拼搏现实的明天。/ 2022年03月11日 10:01/ 0 赞/ 494 阅读