ons fin 整合 連接 shu date pri 對數 data

整合Kafka+Storm，消息通過各種方式進入到Kafka消息中間件，比如通過使用Flume來收集的日誌數據，然後暫由Kafka中的路由暫存，然後在由實時計算程序Storm做實時分析，這時候我們需要講Storm中的Spout中讀取Kafka中的消息，然後交由具體的Bolt組件分析處理。實際上在 apache-storm-0.9.3這個版本的Storm已經自帶了一個集成Kafka的外部插件程序storm-kafka，可以直接使用，例如我使用的Maven依賴配置。

1、配置Maven依賴包 [html] view plain copy

1. <dependency>
2. <groupId>org.apache.kafka</groupId>
3. <artifactId>kafka_2.10</artifactId>
4. <version>0.8.2.0</version>
5. <exclusions>
6. <exclusion>
7. <groupId>org.slf4j</groupId>
8. <artifactId>slf4j-log4j12</artifactId>
9. </exclusion>
10. </exclusions>
11. </dependency>
13. <!-- kafka整合storm -->
14. <dependency>
15. <groupId>org.apache.storm</groupId>
16. <artifactId>storm-core</artifactId>
17. <version>0.9.3</version>
18. <scope>provided</scope>
19. <exclusions>
20. <exclusion>
21. <groupId>org.slf4j</groupId>
22. <artifactId>log4j-over-slf4j</artifactId>
23. </exclusion>
24. <exclusion>
25. <groupId>org.slf4j</groupId>
26. <artifactId>slf4j-api</artifactId>
27. </exclusion>
28. </exclusions>
29. </dependency>
31. <dependency>
32. <groupId>org.apache.storm</groupId>
33. <artifactId>storm-kafka</artifactId>
34. <version>0.9.3</version>
35. </dependency>

storm程序能接收到數據，並進行處理，但是會發現數據被重復處理這是因為在bolt中沒有對數據進行確認，需要調用ack或者fail方法， 修改完成之後即可。 2、編寫Storm程序 [java] view plain copy

1. package com.yun.storm;
2. import java.util.UUID;
4. import storm.kafka.KafkaSpout;
5. import storm.kafka.SpoutConfig;
6. import storm.kafka.ZkHosts;
7. import backtype.storm.Config;
8. import backtype.storm.LocalCluster;
9. import backtype.storm.topology.TopologyBuilder;
11. /**
12. * Storm讀取Kafka消息中間件數據
13. *
14. * @author shenfl
15. *
16. */
17. public class KafkaLogProcess {
20. private static final String BOLT_ID = LogFilterBolt.class.getName();
21. private static final String SPOUT_ID = KafkaSpout.class.getName();
23. public static void main(String[] args) {
25. TopologyBuilder builder = new TopologyBuilder();
26. //表示kafka使用的zookeeper的地址
27. String brokerZkStr = "192.168.2.20:2181";
28. ZkHosts zkHosts = new ZkHosts(brokerZkStr);
29. //表示的是kafak中存儲數據的主題名稱
30. String topic = "mytopic";
31. //指定zookeeper中的一個根目錄，裏面存儲kafkaspout讀取數據的位置等信息
32. String zkRoot = "/kafkaspout";
33. String id = UUID.randomUUID().toString();
34. SpoutConfig spoutconf = new SpoutConfig(zkHosts, topic, zkRoot, id);
35. builder.setSpout(SPOUT_ID , new KafkaSpout(spoutconf));
36. builder.setBolt(BOLT_ID,new LogFilterBolt()).shuffleGrouping(SPOUT_ID);
38. LocalCluster localCluster = new LocalCluster();
39. localCluster.submitTopology(KafkaLogProcess.class.getSimpleName(), new Config(),builder.createTopology() );
40. }
41. }

[java] view plain copy

1. package com.yun.storm;
3. import java.util.Map;
4. import java.util.regex.Matcher;
5. import java.util.regex.Pattern;
7. import backtype.storm.task.OutputCollector;
8. import backtype.storm.task.TopologyContext;
9. import backtype.storm.topology.OutputFieldsDeclarer;
10. import backtype.storm.topology.base.BaseRichBolt;
11. import backtype.storm.tuple.Tuple;
13. /**
14. * 處理來自KafkaSpout的tuple，並保存到數據庫中
15. *
16. * @author shenfl
17. *
18. */
19. public class LogFilterBolt extends BaseRichBolt {
21. private OutputCollector collector;
22. /**
23. *
24. */
25. private static final long serialVersionUID = 1L;
27. Pattern p = Pattern.compile("省公司鑒權接口url\\[(.*)]\\,響應時間\\[([0-9]+)\\],當前時間\\[([0-9]+)\\]");
29. /**
30. * 每個LogFilterBolt實例僅初始化一次
31. */
32. @Override
33. public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
34. this.collector = collector;
35. }
37. @Override
38. public void execute(Tuple input) {
39. try {
40. // 接收KafkaSpout的數據
41. byte[] bytes = input.getBinaryByField("bytes");
42. String value = new String(bytes).replaceAll("[\n\r]", "");
43. // 解析數據並入庫
44. Matcher m = p.matcher(value);
45. if (m.find()) {
46. String url = m.group(1);
47. String usetime = m.group(2);
48. String currentTime = m.group(3);
49. System.out.println(url + "->" + usetime + "->" + currentTime);
50. }
51. this.collector.ack(input);
52. } catch (Exception e) {
53. this.collector.fail(input);
54. }
55. }
57. @Override
58. public void declareOutputFields(OutputFieldsDeclarer declarer) {
59. }
60. }

3、解析日誌入庫 3.1 引入Maven依賴包 [java] view plain copy

1. <!-- mysql maven相關依賴 -->
2. <dependency>
3. <groupId>commons-dbutils</groupId>
4. <artifactId>commons-dbutils</artifactId>
5. <version>1.6</version>
6. </dependency>
7. <dependency>
8. <groupId>mysql</groupId>
9. <artifactId>mysql-connector-java</artifactId>
10. <version>5.1.29</version>
11. </dependency>

3.2 編寫MyDbUtils工具類 （1）創建數據表 [sql] view plain copy

1. create database jfyun;
3. CREATE TABLE `log_info` (
4. `id` int(10) NOT NULL AUTO_INCREMENT,
5. `topdomain` varchar(100) COLLATE latin1_german1_ci DEFAULT NULL,
6. `usetime` varchar(10) COLLATE latin1_german1_ci DEFAULT NULL,
7. `time` datetime DEFAULT NULL,
8. PRIMARY KEY (`id`)
9. ) ENGINE=InnoDB AUTO_INCREMENT=1803 DEFAULT CHARSET=latin1 COLLATE=latin1_german1_ci

（2）MyDbUtils的程序 [java] view plain copy

1. package com.yun.storm.util;
3. import java.sql.Connection;
4. import java.sql.DriverManager;
5. import java.sql.ResultSet;
6. import java.sql.SQLException;
7. import java.util.ArrayList;
8. import java.util.Date;
9. import java.util.List;
11. import org.apache.commons.dbutils.BasicRowProcessor;
12. import org.apache.commons.dbutils.QueryRunner;
13. import org.apache.commons.dbutils.handlers.ArrayListHandler;
15. public class MyDbUtils {
17. private static String className = "com.mysql.jdbc.Driver";
18. private static String url = "jdbc:mysql://192.168.2.20:3306/jfyun?useUnicode=true&characterEncoding=utf-8";
19. private static String user = "root";
20. private static String password = "123";
21. private static QueryRunner queryRunner = new QueryRunner();
23. public static final String INSERT_LOG = "insert into log_info(topdomain,usetime,time) values(?,?,?)";
25. static{
26. try {
27. Class.forName(className);
28. } catch (ClassNotFoundException e) {
29. e.printStackTrace();
30. }
31. }
32. public static void main(String[] args) throws Exception {
33. String topdomain = "taobao.com";
34. String usetime = "100";
35. String currentTime="1444218216106";
36. MyDbUtils.update(MyDbUtils.INSERT_LOG, topdomain,usetime,currentTime);
37. update(INSERT_LOG,topdomain,usetime,MyDateUtils.formatDate2(new Date(Long.parseLong(currentTime))));
38. }
39. /**
40. * @param conn
41. * @throws SQLException
42. */
43. public static void update(String sql,Object... params) throws SQLException {
44. Connection connection = getConnection();
45. //更新數據
46. queryRunner.update(connection,sql, params);
47. connection.close();
48. }
50. public static List<String> executeQuerySql(String sql) {
52. List<String> result = new ArrayList<String>();
53. try {
54. List<Object[]> requstList = queryRunner.query(getConnection(), sql,
55. new ArrayListHandler(new BasicRowProcessor() {
56. @Override
57. public <Object> List<Object> toBeanList(ResultSet rs,
58. Class<Object> type) throws SQLException {
59. return super.toBeanList(rs, type);
60. }
61. }));
62. for (Object[] objects : requstList) {
63. result.add(objects[0].toString());
64. }
65. } catch (SQLException e) {
66. e.printStackTrace();
67. }
68. return result;
69. }
70. /**
71. * @throws SQLException
72. *
73. */
74. public static Connection getConnection() throws SQLException {
75. //獲取mysql連接
76. return DriverManager.getConnection(url, user, password);
77. }
78. }

（3）修改storm程序 [java] view plain copy

1. if (m.find()) {
2. url = m.group(1);
3. usetime = m.group(2);
4. currentTime = m.group(3);
5. System.out.println(url + "->" + usetime + "->" + currentTime);
7. MyDbUtils.update(MyDbUtils.INSERT_LOG, url, usetime,
8. MyDateUtils.formatDate2(new Date(Long.parseLong(currentTime))));
9. }

（4）統計指標 [sql] view plain copy

1. --統計每個url平均響應時間
2. SELECT
3. topdomain,
4. ROUND(AVG(usetime) / 1000, 2) avg_use_time
5. FROM
6. log_info
7. GROUP BY topdomain;

原文鏈接：http://blog.csdn.net/shenfuli/article/details/48982687

storm 整合 kafka之保存MySQL數據庫