使用事物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 + "]";

    }

}