Storm集成 JDBC

• 创建 maven 工程，pom 文件如下：

<dependencies>
	<dependency>
		<groupId>org.apache.storm</groupId>
		<artifactId>storm-core</artifactId>
		<version>1.0.3</version>
		<scope>provided</scope>
	</dependency>
	<!-- 与jdbc集成 -->
	<dependency>
		<groupId>org.apache.storm</groupId>
		<artifactId>storm-jdbc</artifactId>
		<version>1.0.3</version>
		<scope>provided</scope>
	</dependency>
	<dependency>
		<groupId>mysql</groupId>
		<artifactId>mysql-connector-java</artifactId>
		<version>5.1.43</version>
	</dependency>
</dependencies>

• Spout 任务代码如下：

package storm;

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 static final long serialVersionUID = 1571765705181254611L;

	// 模拟数据
	private String[] data = {"I love Beijing", "I love China", "Beijing is the capital of China"};
	
	// 用于往下一个组件发送消息
	private SpoutOutputCollector collector;
	
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		this.collector = collector;
	}

	public void nextTuple() {
		Utils.sleep(3000);
		// 由Strom框架调用，用于接收外部数据源的数据
		int random = (new Random()).nextInt(3);
		String sentence = data[random];
		
		// 发送数据
		System.out.println("发送数据：" + sentence);
		this.collector.emit(new Values(sentence));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("sentence"));
	}
}

• 用于分词的 Bolt 任务代码如下：

package storm;

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 WordCountSplitBolt extends BaseRichBolt {

	private static final long serialVersionUID = -7399165475264468561L;

	private OutputCollector collector;
	
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	public void execute(Tuple tuple) {
		String sentence = tuple.getStringByField("sentence");
		// 分词
		String[] words = sentence.split(" ");
		for (String word : words) {
			this.collector.emit(new Values(word, 1));
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word", "count"));
	}
}

• 用于计数的 Bolt 任务：

package storm;

import java.util.HashMap;
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 WordCountBoltCount extends BaseRichBolt {

	private static final long serialVersionUID = -3206516572376524950L;

	private OutputCollector collector;
	
	private Map<String, Integer> result = new HashMap<String, Integer>();
	
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	public void execute(Tuple tuple) {
		String word = tuple.getStringByField("word");
		int count = tuple.getIntegerByField("count");
		
		if (result.containsKey(word)) {
			result.put(word, result.get(word) + count);
		} else {
			result.put(word, 1);
		}
		// 直接输出到屏幕
		System.out.println("输出的结果是：" + result);
		
		// 将统计结果插入到数据库中
		this.collector.emit(new Values(word, result.get(word)));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word", "total"));
	}
}

• 用于连接的 ConnectionProvider 的代码如下：

package jdbc;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

import org.apache.storm.jdbc.common.ConnectionProvider;

public class MyConnectionProvider implements ConnectionProvider {

	private static final long serialVersionUID = -4784999115987415445L;

	private static String driver = "com.mysql.jdbc.Driver";
	
	private static String url = "jdbc:mysql://qujianlei:3306/storm";
	
	private static String user = "root";
	
	private static String password = "123";
	
	static {
		try {
			Class.forName(driver);
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}

	public Connection getConnection() {
		try {
			return DriverManager.getConnection(url, user, password);
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return null;
	}

	public void prepare() {
	}
	
	public void cleanup() {
	}
}

• 获取 JdbcBolt 的工具类如下：

package jdbc;

import org.apache.storm.jdbc.bolt.JdbcInsertBolt;
import org.apache.storm.jdbc.common.ConnectionProvider;
import org.apache.storm.jdbc.mapper.JdbcMapper;
import org.apache.storm.jdbc.mapper.SimpleJdbcMapper;
import org.apache.storm.topology.IRichBolt;

public class JdbcBoltUtils {
	public static IRichBolt createJDBCBolt() {
		ConnectionProvider connectionProvider = new MyConnectionProvider();
		JdbcMapper simpleJdbcMapper = new SimpleJdbcMapper("result", connectionProvider);
		return new JdbcInsertBolt(connectionProvider, simpleJdbcMapper).
			withTableName("result").withQueryTimeoutSecs(30);
	}
}

注：result 为表的名字，共有两个字段：word, total

• Topology 的代码如下：

package storm;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import jdbc.JdbcBoltUtils;

public class WordCountTopology {

	public static void main(String[] args) throws Exception {
		TopologyBuilder builder = new TopologyBuilder();
		
		// 设置任务的spout组件
		builder.setSpout("wordcount_spout", new WordCountSpout());
		
		// 设置任务的第一个bolt组件
		builder.setBolt("wordcount_splitbolt", new WordCountSplitBolt()).
			shuffleGrouping("wordcount_spout");
		
		// 设置任务的第二个bolt组件
		builder.setBolt("wordcount_count", new WordCountBoltCount()).
			fieldsGrouping("wordcount_splitbolt", new Fields("word"));
		
		// 设置任务的第三个bolt组件将数据持久化到mysql
		builder.setBolt("wordcount_jdbcBolt", JdbcBoltUtils.createJDBCBolt()).
			shuffleGrouping("wordcount_count");
		
		// 创建Topology任务
		StormTopology wc = builder.createTopology();
		
		Config config = new Config();
		
		// 提交到本地运行
		LocalCluster localCluster = new LocalCluster();
		localCluster.submitTopology("mywordcount", config, wc);
		
		// 提交任务到Storm集群运行
//		StormSubmitter.submitTopology(args[0], config, wc);
	}
}

• 右击，运行即可（注：Eclipse 要以管理员身份启动）。