storm-jdbc的使用

Posted 呢喃的歌声

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了storm-jdbc的使用相关的知识,希望对你有一定的参考价值。

最近项目组分配到研究storm-jdbc用法

发现网上关于insert和query方法挺多的,但是自定义方法很少。而且用法上也挺多缺陷。在此自己总结记录一下

 

JdbcInsertBolt 的核心代码

/*
*
* JdbcInsertBolt org.apache.storm 默认包下 此类不会进行emit。所以插入数据后不能接下层bolt
* 如果需要在插入数据后继续往下游流数据必须在源码中进行添加 在github可以获得类(JdbcLookupBolt也是类似,响应还有hbase和redis)
* */
public static JdbcInsertBolt getJdbcInsertBolt() {
System.out.println("had run.........==============");
//Fields outputFields = new Fields("session","time","count");//输出给下层bolt
//get session time
List<Column> schemaColumns = Lists.newArrayList(new Column("session", Types.VARCHAR),//上游过来的字段
new Column("time", Types.INTEGER));
JdbcMapper simpleJdbcMapper = new SimpleJdbcMapper(schemaColumns);
//SimpleJdbcLookupMapper lookupMapper = new SimpleJdbcLookupMapper(outputFields, schemaColumns);
JdbcInsertBolt jdbcInsertBolt = new JdbcInsertBolt(connectionProvider,simpleJdbcMapper)
.withInsertQuery("insert into webinfo(session,time) values(?,?)")
.withQueryTimeoutSecs(50);
return jdbcInsertBolt;
}

//query的核心代码

public static JdbcLookupBolt getJdbcLookupBlot() {
Fields outputFields = new Fields("session","time","count");//输出给下层bolt
List<Column> queryParamColumns = Lists.newArrayList(new Column("session", Types.VARCHAR),
new Column("time", Types.INTEGER)//上层bolt流入
);
//String insertSQL = "insert into student_infos(name,age) values(?,?)";
String selectSql = "select count(*) count from student_infos where session=? and time=?";
SimpleJdbcLookupMapper lookupMapper = new SimpleJdbcLookupMapper(outputFields, queryParamColumns);
JdbcLookupBolt userNameLookupBolt = new JdbcLookupBolt(connectionProvider, selectSql, lookupMapper)
.withQueryTimeoutSecs(30);

//return userNameLookupBolt;
// JdbcState.Options options = new JdbcState.Options()
// .withConnectionProvider(connectionProvider)
// .withJdbcLookupMapper(new SimpleJdbcLookupMapper(new Fields("user_name"), Lists.newArrayList(new Column("user_id", Types.INTEGER))))
// .withSelectQuery("select user_name from user_details where user_id = ?")
// .withQueryTimeoutSecs(30);
// JdbcStateFactory jdbcStateFactory = new JdbcStateFactory(options);
return userNameLookupBolt;
}

//自定义查询方法的核心代码

private JdbcClient client;
public void setup() {
// Map map = Maps.newHashMap();
// map.put("dataSourceClassName","org.hsqldb.jdbc.JDBCDataSource");//com.mysql.jdbc.jdbc2.optional.MysqlDataSource
// map.put("dataSource.url", "jdbc:hsqldb:mem:test");//jdbc:mysql://localhost/test
// map.put("dataSource.user","SA");//root
// map.put("dataSource.password","");//password
// ConnectionProvider connectionProvider = new HikariCPConnectionProvider(map);
connectionProvider.prepare();

int queryTimeoutSecs = 60;
this.client = new JdbcClient(connectionProvider, queryTimeoutSecs);
client.executeSql("UPDATE webinfo s SET s.session=‘haitao‘ where s.time = ‘4‘ ");
}

关于用stormJDBC 效率提升上还是挺大的,项目组兄弟自己写插入一条在50MS左右,大数据下这种情况是不允许的,太慢,然后通过storm开源工具插入,大概在初始化50MS,后面每条是5MS。当然想更快应该也可以做事务JDBC。暂时还没研究

更多参考资料:

https://insight.io/github.com/apache/storm/tree/HEAD/external/storm-jdbc/

 
















































以上是关于storm-jdbc的使用的主要内容,如果未能解决你的问题,请参考以下文章

秒表的使用方法和技巧(秒表的使用方法)

使用“使用严格”作为“使用强”的备份

在使用加载数据流步骤的猪中,使用(使用 PigStorage)和不使用它有啥区别?

Sqlmap的使用

Kettle java脚本组件的使用说明(简单使用升级使用)

PageHelp的使用getParameter的使用zpage-nav的使用QRcode的使用wxpay微信支付的使用jackson-xml数据的转换