我的storm拓扑既不工作(不生成输出)也不失败(不生成错误或异常)

dzhpxtsq  于 2021-06-21  发布在  Storm
关注(0)|答案(1)|浏览(280)

我有一个拓扑结构，我试图统计simulatorspout（不是真正的流）生成的单词出现次数，在写入mysql数据库表之后，表方案非常简单：

  1. Field  |  Type        |  ...
  2. ID     |  int(11)     |      Auto_icr
  3. word   |  varchar(50) |  
  4. count  |  int(11)     |

但是我遇到了一个奇怪的问题（如前所述），我成功地将拓扑提交给了我的storm集群，它由4个管理者组成，我可以在storm web ui中看到拓扑的流程（没有例外），但是当我检查mysql表时，出乎我的意料，表是空的。。。
欢迎任何意见、建议。。。
下面是喷口和螺栓：

  1. public class MySQLConnection {
  3.     private static Connection conn = null;
  4.     private static String dbUrl = "jdbc:mysql://192.168.0.2:3306/test?";
  5.     private static String dbClass = "com.mysql.jdbc.Driver";
  7.     public static Connection getConnection() throws SQLException, ClassNotFoundException  {
  9.         Class.forName(dbClass);
  10.         conn = DriverManager.getConnection(dbUrl, "root", "qwe123");            
  11.         return conn;    
  12.      }
  13. }

======================================句子喷口===============================

  1. public class SentenceSpout extends BaseRichSpout{
  3. private static final long serialVersionUID = 1L;
  4. private boolean _completed = false;
  6. private SpoutOutputCollector _collector;
  7. private String [] sentences = {
  8.         "Obama delivered a powerfull speech against USA",
  9.         "I like cold beverages",
  10.         "RT http://www.turkeyairline.com Turkish Airlines has delayed some flights",
  11.         "don't have a cow man...",
  12.         "i don't think i like fleas"
  13.     };
  15. private int index = 0;
  17. public void open (Map config, TopologyContext context, SpoutOutputCollector collector) {
  18.     _collector = collector;
  19. }
  21. public void nextTuple () {
  23.         _collector.emit(new Values(sentences[index]));
  24.         index++;
  25.         if (index >= sentences.length) {
  26.             index = 0;  
  28.         Utils.waitForSeconds(1);
  29.     }
  30. }
  32. public void declareOutputFields(OutputFieldsDeclarer declarer) {
  33.     declarer.declare(new Fields("sentence"));
  34. }
  36. public void ack(Object msgId) {
  37.     System.out.println("OK: " + msgId);
  38. }   
  40. public void close() {}
  42. public void fail(Object msgId) {
  43.     System.out.println("FAIL: " + msgId);
  44.     }
  45. }

=====================================分离螺栓==============================

  1. public class SplitSentenceBolt extends BaseRichBolt {
  3. private static final long serialVersionUID = 1L;
  4. private OutputCollector _collector;
  6. public void prepare (Map config, TopologyContext context, OutputCollector collector) {
  7.     _collector = collector;
  8. }
  10. public void execute (Tuple tuple) {
  12.     String sentence = tuple.getStringByField("sentence");
  13.     String httpRegex = "((https?|ftp|telnet|gopher|file)):((//)|(\\\\))+[\\w\\d:#@%/;$()~_?\\+-=\\\\\\.&]*";
  14.     sentence = sentence.replaceAll(httpRegex, "").replaceAll("RT", "").replaceAll("[.|,]", "");
  15.     String[] words = sentence.split(" ");
  16.     for (String word : words) {
  17.         if (!word.isEmpty())
  18.         _collector.emit(new Values(word.trim()));
  19.     }
  20.     _collector.ack(tuple);
  21. }
  23. public void declareOutputFields(OutputFieldsDeclarer declarer) {
  24.     declarer.declare(new Fields("word"));
  25.    }
  26.  }

==================================wordcountbolt=================================

  1. public class WordCountBolt extends BaseRichBolt {
  2. private static final long serialVersionUID = 1L;
  3. private HashMap<String , Integer> counts = null;
  4. private OutputCollector _collector; 
  5. private ResultSet resSet = null;
  6. private Statement stmt = null;
  7. private Connection _conn = null;
  9. private String path = "/home/hduser/logOfStormTops/logger.txt";
  10. String rLine = null;
  12. public void prepare (Map config, TopologyContext context, OutputCollector collector) {
  13.     counts = new HashMap<String, Integer>();
  14.     _collector = collector;
  15. }
  17. public void execute (Tuple tuple) {
  19.     int insertResult = 0;
  20.     int updateResult = 0;
  22.     String word = tuple.getStringByField("word");
  23.     //----------------------------------------------------
  24.     if (!counts.containsKey(word)) {            
  26.         counts.put(word, 1);
  27.         try {
  28.             insertResult = wordInsertIfNoExist(word);
  29.             if (insertResult == 1) {
  30.                 _collector.ack(tuple);
  31.             } else {
  32.                 _collector.fail(tuple);
  33.             }
  34.         } catch (ClassNotFoundException e) {
  35.             e.printStackTrace();
  36.         } catch (SQLException e) {
  37.             e.printStackTrace();
  38.         }
  39.     }  else {
  40.     //-----------------------------------------------
  41.         counts.put(word, counts.get(word) + 1);
  43.         try {
  44.             // writing to db
  45.             updateResult = updateCountOfExistingWord(word);
  46.             if (updateResult == 1) {
  47.                 _collector.ack(tuple);
  48.             } else {
  49.                 _collector.fail(tuple);
  50.             }
  51.             // Writing to file
  52.             BufferedWriter buffer = new BufferedWriter(new FileWriter(path));
  53.             buffer.write("[ " + word + " : " + counts.get("word") + " ]");
  54.             buffer.newLine();
  55.             buffer.flush();
  56.             buffer.close();
  58.         } catch (ClassNotFoundException e) {            
  59.             e.printStackTrace();
  60.         } catch (SQLException e) {          
  61.             e.printStackTrace();
  62.         } catch (IOException e) {
  63.             e.printStackTrace();
  64.         }
  65.         System.out.println("{word-" + word + " : count-" + counts.get(word) + "}");
  66.     }
  68. }
  70. public void declareOutputFields(OutputFieldsDeclarer declarer) {
  71. }
  73. //*****************************************************
  75. public int wordInsertIfNoExist(String word) throws ClassNotFoundException, SQLException {
  77.     String query = "SELECT word FROM wordcount WHERE word=\"" + word + "\"";
  78.     String insert = "INSERT INTO wordcount (word, count) VALUES (\"" + word +  "\", 1)";
  79.     _conn = MySQLConnection.getConnection();
  80.     stmt = _conn.createStatement();
  81.     resSet = stmt.executeQuery(query);
  82.     int res = 0;
  83.     if (!resSet.next()) {
  85.          res = stmt.executeUpdate(insert);
  87.     } else {
  88.         System.out.println("Yangi qiymatni kirityotrganda nimadir sodir bo'ldi");
  89.     }
  90.     resSet.close();
  91.     stmt.close();
  92.     _conn.close();
  93.     return res;
  94. }
  96. public int updateCountOfExistingWord(String word) throws ClassNotFoundException, SQLException {
  98.     String update = "UPDATE wordcount SET count=count+1 WHERE word=\"" + word + "\"";   
  99.     _conn = MySQLConnection.getConnection();        
  100.     stmt = _conn.createStatement();
  101.     int result = stmt.executeUpdate(update);
  103.         //System.out.println(word + "'s count has been updated (incremented)");
  105.     resSet.close();
  106.     stmt.close();
  107.     _conn.close();
  108.     return result;
  109.     } 
  110.   }

=============================字计数拓扑==============================

  1. public class WordCountTopology {
  2. private static final String SENTENCE_SPOUT_ID = "sentence-spout";
  3. private static final String SPLIT_BOLT_ID = "split-bolt";
  4. private static final String COUNT_BOLT_ID = "count-bolt";
  5. private static final String TOPOLOGY_NAME = "NewWordCountTopology";
  7. @SuppressWarnings("static-access")
  8. public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
  10. SentenceSpout spout = new SentenceSpout();
  11. SplitSentenceBolt splitBolt = new SplitSentenceBolt();
  12. WordCountBolt countBolt = new WordCountBolt();
  14. TopologyBuilder builder = new TopologyBuilder();
  16. builder.setSpout(SENTENCE_SPOUT_ID, spout, 2);
  18. builder.setBolt(SPLIT_BOLT_ID, splitBolt, 4).shuffleGrouping(SENTENCE_SPOUT_ID);
  20. builder.setBolt(COUNT_BOLT_ID, countBolt, 4).fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
  22. Config config = new Config();
  23. config.setMaxSpoutPending(100);
  24. config.setDebug(true);
  26. StormSubmitter submitter = new StormSubmitter(); 
  28. submitter.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
  30.    }
  31. }
Javaapache-storm

1条答案

按热度按时间
vom3gejh

vom3gejh1#

这是因为当抛出异常时，没有调用\u collector.ack（tuple）。当挂起的元组太多时，spout将停止发送新元组。尝试抛出runtimeexception而不是printstacktrace。

赞(0）分享  回复(0）举报  2021-06-21
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com