使用事物TridentTopology 持久化数据到MySQL 1、构建拓扑JDBCTopology类
package storm.trident.mysql; import java.util.Arrays;
import java.util.Map; import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.trident.TridentState;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.CombinerAggregator;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.spout.IBatchSpout;
import org.apache.storm.trident.state.StateType;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values; /**
* 事物Trident-MySQL Topology
* @author mengyao
*
*/
@SuppressWarnings("all")
public class JDBCTopology { public static void main(String[] args) {
TridentTopology topology = new TridentTopology(); //Spout数据源
FixedBatchSpout spout = new FixedBatchSpout(new Fields("tels"), 7,
new Values("189111 3"),
new Values("135111 7"),
new Values("189111 2"),
new Values("158111 5"),
new Values("159111 6"),
new Values("159111 3"),
new Values("158111 5")
);
spout.setCycle(false); //State持久化配置属性
JDBCStateConfig config = new JDBCStateConfig();
config.setDriver("com.mysql.jdbc.Driver");
config.setUrl("jdbc:mysql://localhost:3306/test");
config.setUsername("root");
config.setPassword("123456");
config.setBatchSize(10);
config.setCacheSize(10);
config.setType(StateType.TRANSACTIONAL);
config.setCols("tel");
config.setColVals("sum");
config.setTable("tbl_tel"); topology.newStream("spout", spout)
.each(new Fields("tels"), new KeyValueFun(), new Fields("tel", "money"))
.groupBy(new Fields("tel"))
.persistentAggregate(JDBCState.getFactory(config), new Fields("money"), new SumCombinerAgg(), new Fields("sum")); LocalCluster cluster = new LocalCluster();
cluster.submitTopology("test1", new Config(), topology.build());
} } @SuppressWarnings("all")
class KeyValueFun extends BaseFunction {
@Override
public void execute(TridentTuple tuple, TridentCollector collector) {
String record = tuple.getString(0);
collector.emit(new Values(record.split("\t")[0], record.split("\t")[1]));
}
} @SuppressWarnings("all")
class SumCombinerAgg implements CombinerAggregator<Long> {
@Override
public Long init(TridentTuple tuple) {
return Long.parseLong(tuple.getString(0));
}
@Override
public Long combine(Long val1, Long val2) {
Long val = val1+val2;
System.out.println(val);
return val;
}
@Override
public Long zero() {
return 0L;
}
} 2、构建基于IBackingMap的JDBCState类
package storm.trident.mysql; import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map; import org.apache.storm.task.IMetricsContext;
import org.apache.storm.trident.state.OpaqueValue;
import org.apache.storm.trident.state.State;
import org.apache.storm.trident.state.StateFactory;
import org.apache.storm.trident.state.StateType;
import org.apache.storm.trident.state.TransactionalValue;
import org.apache.storm.trident.state.map.CachedMap;
import org.apache.storm.trident.state.map.IBackingMap;
import org.apache.storm.trident.state.map.NonTransactionalMap;
import org.apache.storm.trident.state.map.OpaqueMap;
import org.apache.storm.trident.state.map.TransactionalMap; @SuppressWarnings("all")
public class JDBCState<T> implements IBackingMap<T> { private static JDBCStateConfig config; JDBCState(JDBCStateConfig config){
this.config = config;
} @Override
public List<T> multiGet(List<List<Object>> keys) {
StringBuilder sqlBuilder = new StringBuilder("SELECT ").append(config.getCols())
.append(","+config.getColVals())
.append(",txid")
.append(" FROM "+config.getTable())
.append(" WHERE ")
.append(config.getCols())
.append("='"); JDBCUtil jdbcUtil = new JDBCUtil(config.getDriver(),config.getUrl(),config.getUsername(),config.getPassword()); List<Object> result = new ArrayList<Object>();
Map<String, Object> map = null;
for (List<Object> list : keys) {
Object key = list.get(0);
map = jdbcUtil.queryForMap(sqlBuilder.toString()+key+"'");
System.out.println(sqlBuilder.toString()+key+"'"+" 【"+map);
Bean itemBean = (Bean)map.get(key);
long txid=0L;
long val=0L;
if (itemBean!=null) {
val=itemBean.getSum();
txid=itemBean.getTxid();
}
if (config.getType()==StateType.OPAQUE) {
result.add(new OpaqueValue(txid, val));
} else if (config.getType()==StateType.TRANSACTIONAL) {
result.add(new TransactionalValue(txid, val));
} else {
result.add(val);
}
}
return (List<T>) result;
} @Override
public void multiPut(List<List<Object>> keys, List<T> vals) {
//构建新增SQL
StringBuilder sqlBuilder = new StringBuilder("INSERT INTO ").append(config.getTable())
.append("("+config.getCols())
.append(","+config.getColVals())
.append(",txid")
.append(",time")
.append(") VALUES ");
for (int i = 0; i < keys.size(); i++) {
List<Object> key = keys.get(i);
if (config.getType()==StateType.TRANSACTIONAL) {
TransactionalValue val = (TransactionalValue)vals.get(i);
sqlBuilder.append("(");
sqlBuilder.append(key.get(0));
sqlBuilder.append(",");
sqlBuilder.append(val.getVal());
sqlBuilder.append(",");
sqlBuilder.append(val.getTxid());
sqlBuilder.append(",NOW()");
sqlBuilder.append("),");
}
}
sqlBuilder.setLength(sqlBuilder.length()-1);
System.out.println(sqlBuilder.toString());
//新增数据
JDBCUtil jdbcUtil = new JDBCUtil(config.getDriver(),config.getUrl(),config.getUsername(),config.getPassword());
jdbcUtil.insert(sqlBuilder.toString());
} public static Factory getFactory(JDBCStateConfig config) {
return new Factory(config);
} static class Factory implements StateFactory {
private static JDBCStateConfig config;
public Factory(JDBCStateConfig config) {
this.config = config;
}
@Override
public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
final CachedMap map = new CachedMap(new JDBCState(config), config.getCacheSize());
System.out.println(config);
if(config.getType()==StateType.OPAQUE) {
return OpaqueMap.build(map);
} else if(config.getType()==StateType.TRANSACTIONAL){
return TransactionalMap.build(map);
}else {
return NonTransactionalMap.build(map);
}
}
} } 3、构建基于IBackingMap的JDBCStateConfig配置类
package storm.trident.mysql; import java.util.List; import org.apache.storm.trident.state.StateType; @SuppressWarnings("all")
public class JDBCStateConfig { private String url;
private String driver;
private String username;
private String password;
private String table;
private int batchSize;
private String cols;
private String colVals;
private int cacheSize = 100;
private StateType type = StateType.OPAQUE; public String getUrl() {
return url;
} public void setUrl(String url) {
this.url = url;
} public String getDriver() {
return driver;
} public void setDriver(String driver) {
this.driver = driver;
} public String getUsername() {
return username;
} public void setUsername(String username) {
this.username = username;
} public String getPassword() {
return password;
} public void setPassword(String password) {
this.password = password;
} public String getTable() {
return table;
} public void setTable(String table) {
this.table = table;
} public int getBatchSize() {
return batchSize;
} public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
} public String getCols() {
return cols;
} public void setCols(String cols) {
this.cols = cols;
} public String getColVals() {
return colVals;
} public void setColVals(String colVals) {
this.colVals = colVals;
} public int getCacheSize() {
return cacheSize;
} public void setCacheSize(int cacheSize) {
this.cacheSize = cacheSize;
} public StateType getType() {
return type;
} public void setType(StateType type) {
this.type = type;
} @Override
public String toString() {
return "Test2StateConfig [url=" + url + ", driver=" + driver + ", username=" + username + ", password="
+ password + ", table=" + table + ", batchSize=" + batchSize + ", cols=" + cols
+ ", colVals=" + colVals + ", cacheSize=" + cacheSize + ", type=" + type + "]";
} } 4、构建JDBC工具类和实体Bean
package storm.trident.mysql; import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map; public class JDBCUtil { private String driver;
private String url;
private String username;
private String password;
private Connection connection;
private PreparedStatement ps;
private ResultSet rs; public JDBCUtil(String driver, String url, String username, String password) {
this.driver = driver;
this.url = url;
this.username = username;
this.password = password;
init();
} void init(){
try {
Class.forName(driver);
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
} public boolean insert(String sql) {
int state = 0;
try {
connection = DriverManager.getConnection(url, username, password);
ps = connection.prepareStatement(sql);
state = ps.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
} finally {
try {
ps.close();
connection.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
if (state>0) {
return true;
}
return false;
} public Map<String, Object> queryForMap(String sql) {
Map<String, Object> result = new HashMap<String, Object>();
try {
connection = DriverManager.getConnection(url, username, password);
ps = connection.prepareStatement(sql);
rs = ps.executeQuery();
if(rs.next()){
Bean iteBean=new Bean(rs.getString("tel"), rs.getLong("sum"), rs.getLong("txid"), null);
result.put(rs.getString("tel"), iteBean);
}
} catch (SQLException e) {
e.printStackTrace();
} finally {
try {
ps.close();
connection.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
return result;
} public String getDriver() {
return driver;
} public void setDriver(String driver) {
this.driver = driver;
} public String getUrl() {
return url;
} public void setUrl(String url) {
this.url = url;
} public String getUsername() {
return username;
} public void setUsername(String username) {
this.username = username;
} public String getPassword() {
return password;
} public void setPassword(String password) {
this.password = password;
} } package storm.trident.mysql; public class Bean { private String tel;
private long sum;
private long txid;
private String time; public Bean(){
} public Bean(String tel, long sum, long txid, String time) {
super();
this.tel = tel;
this.sum = sum;
this.txid = txid;
this.time = time;
} public String getTel() {
return tel;
} public void setTel(String tel) {
this.tel = tel;
} public long getSum() {
return sum;
} public void setSum(long sum) {
this.sum = sum;
} public long getTxid() {
return txid;
} public void setTxid(long txid) {
this.txid = txid;
} public String getTime() {
return time;
} public void setTime(String time) {
this.time = time;
} @Override
public String toString() {
return "Bean [tel=" + tel + ", sum=" + sum + ", txid=" + txid + ", time=" + time + "]";
} }