Strom JDBC 整合学习笔记

原创编写： 王宇 
2016-11-24

---

 

Strom JDBC 整合学习笔记
插入数据库(Inserting into a database)
ConnectionProvider
JdbcMapper
JdbcInsertBolt
SimpleJdbcMapper
JdbcTridentState
查询数据库(Lookup from Database)
SimpleJdbcLookupMapper
JdbcLookupBolt
JdbcTridentState for Lookup
例子
maven中加入storm-jdbc和mysql的connector
表结构
Github中的例子
实例代码

 

---

插入数据库(Inserting into a database)

ConnectionProvider

统一接口 
org.apache.storm.jdbc.common.ConnectionProvider

1. publicinterfaceConnectionProviderextendsSerializable{
2. /**
3. * method must be idempotent.
4. */
5. void prepare();
7. /**
8. *
9. * @return a DB connection over which the queries can be executed.
10. */
11. Connection getConnection();
13. /**
14. * called once when the system is shutting down, should be idempotent.
15. */
16. void cleanup();
17. }

支持： HikariCP 连接池 
org.apache.storm.jdbc.common.HikariCPConnectionProvider

JdbcMapper

org.apache.storm.jdbc.mapper.JdbcMapper

1. publicinterfaceJdbcMapperextendsSerializable{
2. List<Column> getColumns(ITuple tuple);
3. }

getColume方法，定义了一个storm tuple 如何映射一个数据库表的列。 

The order of the returned list is important. The place holders in the supplied queries are resolved in the same order as returned list. 

返回的列表的顺序性是很重要的。总之就是自己在插入数据的时候，各个字段的顺序要对应一致。 
例如：我们提交了一个插入查询语句insert into user(user_id, user_name, create_date) values (?,?, now())那么values（）中第一个？就对应了user_id,第二个？对应了usr_name以此类推。getColumns的返回列表也是这样。咱们的jdbc不提供任何不标准的查询语法。

JdbcInsertBolt

• 要使用JdbcInsertBolt，我们需要用一个ConnectionProvider的实现以及JdbcMapper的实现（该实现将tuple转换成DB的行）来构建一个JdbcInsertBolt实例。
• 用withTableName方法提供表名
• withInsertQuery方法提供一个插入查询
• 设置一个查询超时时间来规定一个查询最多能花多少时间。默认与topology.message.timeout.secs一样大，如果该值为-1那就意味着不设置查询超时。我们可以设置查询超时时间<=topology.message.timeout.secs。

1. Map hikariConfigMap =Maps.newHashMap();
2. hikariConfigMap.put("dataSourceClassName","com.mysql.jdbc.jdbc2.optional.MysqlDataSource");
3. hikariConfigMap.put("dataSource.url","jdbc:mysql://localhost/test");
4. hikariConfigMap.put("dataSource.user","root");
5. hikariConfigMap.put("dataSource.password","password");
6. ConnectionProvider connectionProvider =newHikariCPConnectionProvider(hikariConfigMap);
8. String tableName ="user_details";
9. JdbcMapper simpleJdbcMapper =newSimpleJdbcMapper(tableName, connectionProvider);
11. JdbcInsertBolt userPersistanceBolt =newJdbcInsertBolt(connectionProvider, simpleJdbcMapper)
12. .withTableName("user")
13. .withQueryTimeoutSecs(30);
14. Or
15. JdbcInsertBolt userPersistanceBolt =newJdbcInsertBolt(connectionProvider, simpleJdbcMapper)
16. .withInsertQuery("insert into user values (?,?)")
17. .withQueryTimeoutSecs(30);

SimpleJdbcMapper

更一般化的JdbcMapper，它可以将tuple与数据库的行进行映射。SimpleJdbcMapper假设tuple字段名与你将要写入的数据库表中的列名是一样的。

JdbcTridentState

We also support a trident persistent state that can be used with trident topologies.

1. JdbcState.Options options =newJdbcState.Options()
2. .withConnectionProvider(connectionProvider)
3. .withMapper(jdbcMapper)
4. .withTableName("user_details")
5. .withQueryTimeoutSecs(30);
6. JdbcStateFactory jdbcStateFactory =newJdbcStateFactory(options);

可以使用 withInsertQuery 设置条件

查询数据库(Lookup from Database)

org.apache.storm.jdbc.mapper.JdbcLookupMapper

1. void declareOutputFields(OutputFieldsDeclarer declarer);
2. List<Column> getColumns(ITuple tuple);
3. List<Values> toTuple(ITuple input,List<Column> columns);

• declareOutputFields 
  指定输出的tuple中的字段
• getColumns 
  确定查询中的占位符(?)以及它们的SQL类型和值。
• toTuple 
  接收一个输入tuple并且表示数据库一行的列字段值列表作为select搜索的结果。

SimpleJdbcLookupMapper

针对单表简单查询

• SimpleJdbcMapper认为tuple中的字段与你作为占位符的字段名是一致的

1. Fields outputFields =newFields("user_id","user_name","create_date");
2. List<Column> queryParamColumns =Lists.newArrayList(newColumn("user_id",Types.INTEGER));
3. this.jdbcLookupMapper =newSimpleJdbcLookupMapper(outputFields, queryParamColumns);

JdbcLookupBolt

注意超时设置要<=topology.message.timeout.secs

1. String selectSql ="select user_name from user_details where user_id = ?";
2. SimpleJdbcLookupMapper lookupMapper =newSimpleJdbcLookupMapper(outputFields, queryParamColumns)
3. JdbcLookupBolt userNameLookupBolt =newJdbcLookupBolt(connectionProvider, selectSql, lookupMapper)
4. .withQueryTimeoutSecs(30);

JdbcTridentState for Lookup

1. JdbcState.Options options =newJdbcState.Options()
2. .withConnectionProvider(connectionProvider)
3. .withJdbcLookupMapper(newSimpleJdbcLookupMapper(newFields("user_name"),Lists.newArrayList(newColumn("user_id",Types.INTEGER))))
4. .withSelectQuery("select user_name from user_details where user_id = ?");
5. .withQueryTimeoutSecs(30);

例子

maven中加入storm-jdbc和mysql的connector

1. <dependency>
2. <groupId>org.apache.storm</groupId>
3. <artifactId>storm-jdbc</artifactId>
4. <version>0.10.0</version>
5. <scope>provided</scope>
6. </dependency>
7. <dependency>
8. <groupId>mysql</groupId>
9. <artifactId>mysql-connector-Java</artifactId>
10. <version>5.1.31</version>
11. </dependency>

表结构

1. CREATE TABLE `userinfo`(
2. `id`int(11) NOT NULL AUTO_INCREMENT,
3. `user_id` varchar(50) DEFAULT NULL,
4. `resource_id` varchar(10) DEFAULT NULL,
5. `create_date` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
6. `count`int(11) DEFAULT NULL,
7. PRIMARY KEY (`id`)
8. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

Github中的例子

storm/external/storm-jdbc/src/test/java/org/apache/storm/jdbc/bolt/ 
storm/examples/storm-jdbc-examples/src/main/java/org/apache/storm/jdbc/

实例代码

 

1. import java.sql.Types;
2. import java.util.HashMap;
3. import java.util.List;
4. import java.util.Map;
6. import org.apache.storm.guava.collect.Lists;
7. import org.apache.storm.jdbc.bolt.JdbcInsertBolt;
8. import org.apache.storm.jdbc.bolt.JdbcLookupBolt;
9. import org.apache.storm.jdbc.common.Column;
10. import org.apache.storm.jdbc.common.ConnectionProvider;
11. import org.apache.storm.jdbc.common.HikariCPConnectionProvider;
12. import org.apache.storm.jdbc.mapper.JdbcMapper;
13. import org.apache.storm.jdbc.mapper.SimpleJdbcLookupMapper;
14. import org.apache.storm.jdbc.mapper.SimpleJdbcMapper;
16. import backtype.storm.tuple.Fields;
18. publicclassPersistentBolt{
19. privatestaticMap<String,Object> hikariConfigMap =newHashMap<String,Object>(){{
20. put("dataSourceClassName","com.mysql.jdbc.jdbc2.optional.MysqlDataSource");
21. put("dataSource.url","jdbc:mysql://localhost/storm");
22. put("dataSource.user","user");
23. put("dataSource.password","password");
24. }};
25. publicstaticConnectionProvider connectionProvider =newHikariCPConnectionProvider(hikariConfigMap);
27. publicstaticJdbcInsertBolt getJdbcInsertBolt(){
28. //使用tablename进行插入数据，需要指定表中的所有字段
29. /*String tableName="userinfo";
30. JdbcMapper simpleJdbcMapper = new SimpleJdbcMapper(tableName, connectionProvider);
31. JdbcInsertBolt jdbcInsertBolt = new JdbcInsertBolt(connectionProvider, simpleJdbcMapper)
32. .withTableName("userinfo")
33. .withQueryTimeoutSecs(50);*/
34. //使用schemaColumns，可以指定字段要插入的字段
35. List<Column> schemaColumns =Lists.newArrayList(newColumn("user_id",Types.VARCHAR),
36. newColumn("resource_id",Types.VARCHAR),newColumn("count",Types.INTEGER));
37. JdbcMapper simpleJdbcMapper =newSimpleJdbcMapper(schemaColumns);
38. JdbcInsertBolt jdbcInsertBolt =newJdbcInsertBolt(connectionProvider,simpleJdbcMapper)
39. .withInsertQuery("insert into userinfo(id,user_id,resource_id,count) values(?,?,?)")
40. .withQueryTimeoutSecs(50);
41. return jdbcInsertBolt;
42. }
44. publicstaticJdbcLookupBolt getJdbcLookupBlot(){
45. //查询
46. //指定bolt的输出字段
47. Fields outputFields =newFields("user_id","resource_id","count");
48. //指定查询条件字段
49. List<Column> queryColumns =Lists.newArrayList(newColumn("user_id",Types.VARCHAR),newColumn("resource_id",Types.VARCHAR));
50. String selectSql ="select count from userinfo where user_id=? and resource_id=?";
51. SimpleJdbcLookupMapper lookupMapper =newSimpleJdbcLookupMapper(outputFields, queryColumns);
52. JdbcLookupBolt jdbcLookupBolt =newJdbcLookupBolt(connectionProvider, selectSql, lookupMapper);
53. return jdbcLookupBolt;
54. }
55. }