storm spout mysql_storm+mysql集成
使用storm自带的JdbcInsertBolt插入数据
maven
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4.0.0
com.sid.bigdata
storm
0.0.1
jar
storm
http://maven.apache.org
UTF-8
1.1.1
org.apache.storm
storm-core
${storm.version}
org.apache.storm
storm-jdbc
${storm.version}
mysql
mysql-connector-java
5.1.31
mysql
CREATE TABLE `wordcount` (
`word` varchar(100) DEFAULT NULL,
`count` int(11) DEFAULT NULL
) ENGINE=MyISAM DEFAULT CHARSET=utf8;
spout
package com.neusoft;
import java.util.Map;
import java.util.Random;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
public class WordCountSpout extends BaseRichSpout{
private SpoutOutputCollector collector;
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
public static final String[] words = new String[]{"aaa","bbb","ccc","aa","bb","a"};
/**
* 1.把每一行数据发射出去
* */
public void nextTuple() {
Random random = new Random();
String word =words[random.nextInt(words.length)]; //获取文件中的每行内容
//发射出去
this.collector.emit(new Values(word));
System.out.println("emit: "+word);
Utils.sleep(50);
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}
CountBolt
package com.neusoft;
/**
* Created by ttc on 2018/9/28.
*/
import java.util.Map;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
public class CountBolt extends BaseRichBolt{
private OutputCollector collector;
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
/**
* 业务逻辑
* */
public void execute(Tuple input) {
collector.emit(new Values(input.getStringByField("word"),1));
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
//后面jdbc insert bolt直接把这里的输出写Mysql里去了,所以这里的fileds的名字要跟mysql表的字段名字对应
declarer.declare(new Fields("word","count"));
}
}
LocalWordCountStormJdbcTopology
package com.neusoft;
import java.util.Map;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.jdbc.bolt.JdbcInsertBolt;
import org.apache.storm.jdbc.common.ConnectionProvider;
import org.apache.storm.jdbc.common.HikariCPConnectionProvider;
import org.apache.storm.jdbc.mapper.JdbcMapper;
import org.apache.storm.jdbc.mapper.SimpleJdbcMapper;
import org.apache.storm.shade.com.google.common.collect.Maps;
import org.apache.storm.topology.TopologyBuilder;
public class LocalWordCountStormJdbcTopology {
public static void main(String[] args) {
//本地模式,没有提交到服务器集群上,不需要搭建storm集群
LocalCluster cluster = new LocalCluster();
//TopologyBuilder根据spout和bolt来构建Topology
//storm中任何一个作业都是通过Topology方式进行提交的
//Topology中需要指定spout和bolt的执行顺序
TopologyBuilder tb = new TopologyBuilder();
tb.setSpout("DataSourceSpout", new WordCountSpout());
//SumBolt以随机分组的方式从DataSourceSpout中接收数据
tb.setBolt("CountBolt", new CountBolt()).shuffleGrouping("DataSourceSpout");
Map hikariConfigMap = Maps.newHashMap();
hikariConfigMap.put("dataSourceClassName","com.mysql.jdbc.jdbc2.optional.MysqlDataSource");
hikariConfigMap.put("dataSource.url", "jdbc:mysql://localhost/scott");
hikariConfigMap.put("dataSource.user","root");
hikariConfigMap.put("dataSource.password","root");
ConnectionProvider connectionProvider = new HikariCPConnectionProvider(hikariConfigMap);
//mysql的表名
String tableName = "wordcount";
JdbcMapper simpleJdbcMapper = new SimpleJdbcMapper(tableName, connectionProvider);
JdbcInsertBolt userPersistanceBolt = new JdbcInsertBolt(connectionProvider, simpleJdbcMapper)
.withTableName(tableName)
.withQueryTimeoutSecs(30);
tb.setBolt("JdbcInsertBolt", userPersistanceBolt).shuffleGrouping("CountBolt");
//第一个参数是topology的名称,第三个参数是Topology
cluster.submitTopology("LocalWordCountStormJdbcTopology", new Config(), tb.createTopology());
}
}
storm spout mysql_storm+mysql集成相关推荐
- storm spout mysql_Storm之详解spout、blot
1.Topology的构造 backtype.storm.topology.TopologyBuilder 2.Spout组件的编写 实现接口 backtype.storm.topology.IRic ...
- Storm精华问答 | Storm如何连接MySQL?
Storm是Twitter开源的分布式实时大数据处理框架,被业界称为实时版Hadoop.随着越来越多的场景对Hadoop的MapReduce高延迟无法容忍,大数据实时处理解决方案的应用日趋广泛,目前已 ...
- 强烈推荐一款完全免费的绿色JRE+Tomcat+MySQL集成开发工具 - JTM
为什么80%的码农都做不了架构师?>>> 因为工作需要经常要将JSP项目拿到客户电脑上进行演示,但客户经常在安装配置JDK.Tomcat.MySQL的过程中出现很多问题,给客户 ...
- Spring MVC Hibernate MySQL集成CRUD示例教程
Spring MVC Hibernate MySQL集成CRUD示例教程 我们在上一篇教程中学习了如何集成Spring和Hibernate.今天,我们将继续前进,并将Spring MVC和Hibern ...
- storm spout读取mysql_Storm中Spout使用注意事项小结
Storm中Spout用于读取并向计算拓扑中发送数据源,最近在调试一个topology时遇到了系统qps低,处理速度达不到要求的问题,经过排查后发现是由于对Spout的使用模式不当导致的多线程同步等待 ...
- 通达oa与mysql集成_安装通达OA后想使用其自带的Mysql数据库的解决方法
通达OA是集成了apache/php/mysql的,安装通达OA后,我还想安装别的php+mysql的东西玩玩,但是由于通达人的BT,死活要购买正版后才肯告诉mysql的root密码,没办法,自己动手 ...
- iis php mysql 集成_如何在IIS上集成php(iis+mysql+php+zend)
下面介绍下如何在IIS上集成php. 这里我就不说cgi了,因为cgi需要系统权限过高,不建议虚拟主机使用,而且cgi程序也很少有人用到,楼主说的要iis6.0结合php 安装需要:windows20 ...
- mvc crud_Spring MVC Hibernate MySQL集成CRUD示例教程
mvc crud We learned how to integrate Spring and Hibernate in our last tutorial. Today we will move f ...
- tomcat ng mysql 集成_tomcat ; nginx ;mysql
[说明]上午在服务器上部署了自己的web目录(tomcat虚拟目录),下午安装并了解了nginx(出了点问题,后来解决了),晚上尝试nginx和tomcat配置(一头雾水) 一:今日完成 1)配置to ...
最新文章
- python怎么使用int四舍五入_python中如何取整数
- 中美德工业互联网路径比较
- [hdu6434]Problem I. Count
- python处理csv文件缺失值_Python:如何处理CSV中的缺失值?
- (chap8 确认访问用户身份的认证) 基于表单认证
- [转]python3之模块psutil系统性能信息
- linux make指定目标平台,CMake on Linux:目标平台不支持动态链接
- JAVA中如何取map的值_如何在java中取map中的键值 的两种方法
- js拾遗: replace 替换参数
- CCF2018-3-2 碰撞的小球
- 台式计算机技术参数响应表,联想台式电脑配置推荐及参数详情【图文】
- 企业级实际性能测试案例与经验分享
- 使用Spring框架对接Twitter(二)
- 长隆大马戏机器人_长隆娱乐登陆
- java中静态方法与非静态方法的区别及非静态方法的注意事项
- python贴吧数据可视化软件_Python数据可视化工具之Pyecharts初体验
- Unity获取Android手机的RAM和剩余RAM和ROM
- nginx 通过路径分发 网站/端口/服务
- Excel高级图表制作④——控件+Offset组合,让图表真正动起来!
- 几种蔬菜帮你降糖 血糖过高的人别错过