package mystorm.wordcount;

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

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


/**
 * storm集成jdbc的 连接提供者
 * @author Administrator
 *
 */

//为jdbcBolt组件提供对应的数据连接
public class MyConnectionProvider implements ConnectionProvider {

    private static String driver = "com.mysql.jdbc.driver";
    private static String url = "jdbc:mysql://192.168.2.1:3306/test";
    private static String user = "root";
    private static String password = "123456";
    
    static{
        
        try {
            Class.forName(driver);
        } catch (ClassNotFoundException e) {
            // TODO Auto-generated catch block
            new ExceptionInInitializerError(e);
//            e.printStackTrace();
        }
        
    }
    
    @Override
    public void cleanup() {
        // TODO Auto-generated method stub

    }

    @Override
    public Connection getConnection() {
        // TODO Auto-generated method stub
        try {
            return DriverManager.getConnection(url,user,password);
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void prepare() {
        // TODO Auto-generated method stub

    }

}

/**
 * 
 * 单词计数的Topology的入口，主程序
 * 
 * @author Administrator
 *
 */
public class WordCountTopology {
    
    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
        TopologyBuilder builder = new TopologyBuilder();
        
        //设置任务的spout组件
        builder.setSpout("wordcount_spout", new WordCountSpout());
        
        //设置任务的第一个bolt组件
        builder.setBolt("wordcount_bolt", new WordCountSplitBolt())
        // (随机分配策略接收spout的任务)
                .shuffleGrouping("wordcount_spout");
        
        //设置第二个bolt组件
        builder.setBolt("wordcount_countbolt", new WordCountBoltCount())
        //(接受第一个bolt组件的数据，按字段进行分组)
                .fieldsGrouping("wordcount_bolt", new Fields("word"));
        
        //设置第三个bolt组件（storm提供的），用与把记录保存到mysql数据库中。
        builder.setBolt("wordcount_jdbcbolt", createJDBCBolt())
                .shuffleGrouping("wordcount_countbolt");
        
        //创建topology的任务
        StormTopology wc = builder.createTopology();
        
        //配置参数信息
        Config conf = new Config();
        
        //1.本地模式提交
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology( "mywordcount", conf, wc);
        
        
        /**
         * 
         * 本地模式运行结果
         * 
         * 
            spout采集的数据是：beijing is the capital of china
            输出的结果：{beijing=1}
            输出的结果：{is=1, beijing=1}
            输出的结果：{the=1, is=1, beijing=1}
            输出的结果：{the=1, capital=1, is=1, beijing=1}
            输出的结果：{the=1, capital=1, of=1, is=1, beijing=1}
            输出的结果：{the=1, capital=1, china=1, of=1, is=1, beijing=1}
            spout采集的数据是：I love beijing
            输出的结果：{the=1, capital=1, china=1, of=1, I=1, is=1, beijing=1}
            输出的结果：{the=1, love=1, capital=1, china=1, of=1, I=1, is=1, beijing=1}
            输出的结果：{the=1, love=1, capital=1, china=1, of=1, I=1, is=1, beijing=2}
            spout采集的数据是：I love china
            输出的结果：{the=1, love=1, capital=1, china=1, of=1, I=2, is=1, beijing=2}
            输出的结果：{the=1, love=2, capital=1, china=1, of=1, I=2, is=1, beijing=2}
            输出的结果：{the=1, love=2, capital=1, china=2, of=1, I=2, is=1, beijing=2}
         * 
         * 
         * 
         */
        
        //2集群模式提交
//        StormSubmitter.submitTopology(args[0], conf, wc);
        
    }