Storm 读取MySQL数据实践

maven依赖包：

<dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.1.1</version>
            <scope>provided</scope>
        </dependency>
                <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>6.0.2</version>
        </dependency>
1
2
3
4
5
6
7
8
9
10
11
12

Spout:

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;

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;


public class MysqlSpout extends BaseRichSpout {
    String url = "jdbc:mysql://192.168.1.177:3306/bigdata";
    String username = "root";
    String password = "*******";

    private ResultSet res;
    private Statement sta;
    private SpoutOutputCollector collector;
    int id = 0;

    public void nextTuple() {
        String str = "";
        try {
            if (res.next()) {
                String username = res.getString(1);
                String school = res.getString(2);
                String device = res.getString(3);
                String logintime = res.getString(4);
                str += username+"\t"+school+"\t"+logintime+"\t"+device;
                collector.emit(new Values(str));
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public void open(Map arg0, TopologyContext topology, SpoutOutputCollector collector) {

        try {
            String driver = "com.mysql.jdbc.Driver";
            Class.forName(driver);
            Connection conn = DriverManager.getConnection("jdbc:mysql://192.168.1.177:3306/bigdata", "root", "chineseall");
            sta = conn.createStatement();
            res = sta.executeQuery("select username,school,devicetype,logintime from device_log");
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }

        this.collector = collector;

    }

    public void declareOutputFields(OutputFieldsDeclarer output) {
        output.declare(new Fields("device"));
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

Bolt1：

import java.util.HashMap;
import java.util.Map;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class Trans extends BaseBasicBolt {
    final Map<String, Sql> map = new HashMap<String, Sql>();
    public void execute(Tuple arg0, BasicOutputCollector output) {
        String str = String.valueOf(arg0.getValueByField("device"));
        String[] arrys = str.split("\t");
        String username = arrys[0];
        String school = arrys[1];
        String time = arrys[2].split(" ")[0];
        String logintime = time.split("-")[0] + time.split("-")[1];
        String device = arrys[3];
        Sql sql = null;
        if (map.containsKey(username + " " + logintime)) {
            sql = map.get(username + " " + logintime);
            sql.setUsername(username);
            sql.setSchool(school);
            sql.setDate(logintime);
            sql.setDevice_type(device);
            if("PC".equals(device)){
                int pc = sql.getPc();
                sql.setPc(pc+1);
            }else if("安卓".equals(device)){
                int android = sql.getAndroid();
                sql.setAndroid(android+1);
            }else if("苹果".equals(device)){
                int apple = sql.getApple();
                sql.setApple(apple+1);
            }
            int sum = sql.getSum();
            sql.setSum(sum+1);
        }else{
            sql = new Sql();
            sql.setUsername(username);
            sql.setSchool(school);
            sql.setDate(logintime);
            sql.setDevice_type(device);
            if("PC".equals(device)){
                sql.setPc(1);
            }else if("安卓".equals(device)){
                sql.setAndroid(1);
            }else if("苹果".equals(device)){
                sql.setApple(1);
            }
            sql.setSum(1);  
        }
        map.put(username + " " + logintime, sql);
        output.emit(new Values(sql));
    }

    public void declareOutputFields(OutputFieldsDeclarer output) {
        output.declare(new Fields("device"));
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

Bolt2：

import org.apache.storm.task.OutputCollector;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;

public class Output extends BaseBasicBolt {
    OutputCollector collector;

    public void execute(Tuple arg0, BasicOutputCollector outputCollector) {
        Sql sql = (Sql)arg0.getValueByField("device");
        System.out.println(sql.getUsername()+"\t"
                +sql.getSchool()+"\t"
                +sql.getDate()+"\t"
                +sql.getPc()+"\t"
                +sql.getAndroid()+"\t"
                +sql.getApple()+"\t"
                +sql.getSum());
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

Job：

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;

public class StartStorm {
    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("1", new MysqlSpout());
        builder.setBolt("2", new Trans()).shuffleGrouping("1");
        builder.setBolt("3", new Output()).shuffleGrouping("2");

        Config config = new Config();
        config.setDebug(false);
        if (args != null && args.length > 0) {
            config.setNumWorkers(2);
            StormSubmitter.submitTopology("mysql", config, builder.createTopology());
        } else {
            LocalCluster local = new LocalCluster();
            local.submitTopology("topo", config, builder.createTopology());

        }

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

builder.setBolt(“2”, new Trans()).shuffleGrouping(“1”);
shuffleGrouping(“1”)在这里1记录bolt数据的来源是哪个Spout或者Bolt
builder.setBolt(“3”, new wordcount(), 5).fieldsGrouping(“2”, new Fields(“word”));
第二个参数new Fields(“word”) 表示按照名为word的分组来分发数据