如何使用storm更新状态-(将结果存储在数据库中)

9q78igpj  于 2021-06-24  发布在  Storm
关注(0)|答案(0)|浏览(265)

我在拓扑中的更新状态（本地模式）有问题-我想在数据库中写入结果：
按以下方式运行拓扑：

TridentTopology topology = new TridentTopology();

StateFactory stateFactoryDB = new LocationDBFactory();
MessageSubmittedOpaqueTransSpout spout = new MessageSubmittedOpaqueTransSpout();

Stream globalSubmittedMessageStream = topology.newStream("globalSubmittedMessageStream", spout).parallelismHint(1);

globalSubmittedMessageStream
.project(new Fields("stormTxid","bookingId", "timestamp", "messagesSubmitted", "messagesFailed"))
    .each(new Fields("bookingId"), new FilterNotZero() )
    .each(new Fields("timestamp"), new FilterNull())
    .each(new Fields("bookingId", "messagesSubmitted"),new PrintFilter("success submitted - start: "))
    .partitionPersist(
        stateFactoryDB, 
        new Fields("stormTxid","bookingId", "timestamp", "messagesFailed", "messagesSubmitted"), 
        new LocationDBStateUpdater(),
        new Fields("test")
    );

数据正确地从数据库中获取（通过messagesubmittedopaquetranspout），并由每个过滤器发出，只有来自第一个事务的数据通过partitionpersist保存到statefactorydb（locationdbfactory），非其他事务才会被更新。
locationdbstateupdater类中updatestate方法的代码：

public void updateState(LocationDB state, List<TridentTuple> tuples,TridentCollector collector) {
 for(TridentTuple tuple : tuples){
  LOG.debug("^%*&^$&^#&%$@#^%$#^%$*&^%(&^%(&^%(&%(&%&  > {} ", tuple);
 }
 state.save(tuples, collector);
}

locationdb类的代码：

public void beginCommit(Long txid) {
 LOG.info("beginCommit...... txid: {}", txid);
}

@Override
public void commit(Long txid) {
 LOG.info("commit...... txid: {}", txid);
}

public void save(List<TridentTuple> tuples, TridentCollector collector) {
 LOG.info("save ......... START");      
 for(TridentTuple tuple : tuples){
  LOG.info("#@#@@@@@@@@@@@@@@@@@@@@@@@@@ TEST TUPLE: {}", tuple);
 }      
 for(TridentTuple t : tuples){
  collector.emit(new Values("TRUE"));
 }
 LOG.info("save ......... END");
}

日志：

6056 [Thread-10] INFO  backtype.storm.daemon.worker - Worker f818dc1b-0f34-43f8-b70a-a12afaef9d88 for storm test-topology-1-1373534952 on cbe0de9a-4cef-483f-bc11-77a15b0ad6f9:4 has finished loading 
6073 [Thread-23] INFO  c.n.c.f.imps.CuratorFrameworkImpl - Starting 
0    [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2000] WARN  org.apache.zookeeper.server.NIOServerCnxn  - EndOfStreamException: Unable to read additional data from client sessionid 0x13fcd101957000d, likely client has closed socket
6098 [Thread-23] INFO  backtype.storm.daemon.executor - Prepared bolt spout0:(5) 
6098 [Thread-17] INFO  backtype.storm.daemon.executor - Prepared bolt $spoutcoord-spout0:(2) 
6106 [Thread-25] INFO  c.n.c.f.imps.CuratorFrameworkImpl - Starting 
32   [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2000] WARN  org.apache.zookeeper.server.NIOServerCnxn  - EndOfStreamException: Unable to read additional data from client sessionid 0x13fcd101957000f, likely client has closed socket
6120 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - getCoordinator !!! 
6120 [Thread-25] INFO  backtype.storm.daemon.executor - Opened spout $mastercoord-bg0:(1) 
6122 [Thread-25] INFO  backtype.storm.daemon.executor - Activating spout $mastercoord-bg0:(1) 
6122 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 1 !!! 
6225 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Opening JDBC Connection 
6237 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 2 !!! 
6288 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 3 !!! 
6341 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 4 !!! 
6460 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 5 !!! 
6511 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 6 !!! 
6562 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 7 !!! 
6670 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 8 !!! 
6725 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 9 !!! 
6776 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 10 !!! 
6828 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 11 !!! 
6879 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 12 !!! 
6931 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 13 !!! 
6982 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 14 !!! 
7034 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 15 !!! 
7085 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 16 !!! 
7136 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 17 !!! 
7187 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 18 !!! 
7238 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 19 !!! 
7289 [Thread-25] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - Coordinator - isReady method, txid: 20 !!! 
8238 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Created connection 19739814. 
8391 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ooo Using Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
8392 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==>  Preparing: SELECT MAX(id) as max_id FROM msisdn_status  
8446 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==> Parameters:  
8619 [Thread-23] INFO  c.archermobile.storm.StatusManager - Max id: 2902288 
8620 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Resetting autocommit to true on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
8773 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Closing JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
8773 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Returned connection 19739814 to pool. 
8774 [Thread-23] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - emitPartitionBatch, startId: 0, endId: 50, max id: 2902288, TransactionAttempt: 1:0, partition: 0  
8779 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Opening JDBC Connection 
8779 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Checked out connection 19739814 from pool. 
8780 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Setting autocommit to false on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
8931 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ooo Using Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
8931 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ==>  Preparing: SELECT sum(f.messages_submitted) as messages_submitted, sum(f.messages_failed) as messages_failed, fb.booking_id,f.classification_id,f.submission_account_id, f.timestamp FROM ( SELECT count(IF(msisdn_status.status >= 0, 1, NULL)) as messages_submitted, count(IF(msisdn_status.status < 0, 1, NULL)) as messages_failed, msisdn_status.flight_id, date(msisdn_status.timestamp) as timestamp, msisdn_status.classification_id, msisdn_status.submission_account_id FROM msisdn_status WHERE ( msisdn_status.id > ? AND msisdn_status.id <= ? AND msisdn_status.timestamp IS NOT NULL) GROUP BY flight_id, timestamp, classification_id, submission_account_id ) f LEFT JOIN jupiter.flight_reference fr on f.flight_id = fr.flight_id LEFT JOIN jupiterbs.flightboard fb on fb.flight_reference = fr.flight_reference GROUP BY f.timestamp, fb.booking_id, f.classification_id, f.submission_account_id;  
8932 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ==> Parameters: 0(Long), 50(Long) 
9141 [Thread-23] INFO  c.archermobile.storm.StatusManager -  Submitted List [MsisdnStatus{29, 2011-01-27 00:00:00.0, -1, -1, 6, 0}, MsisdnStatus{29, 2011-01-27 00:00:00.0, 1, -1, 0, 7}, MsisdnStatus{29, 2011-01-27 00:00:00.0, 2, -1, 0, 2}, MsisdnStatus{29, 2011-01-28 00:00:00.0, -1, -1, 23, 0}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 1, -1, 0, 8}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 2, -1, 0, 4}] 
9141 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Resetting autocommit to true on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9292 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Closing JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9292 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Returned connection 19739814 to pool. 
9295 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 0]  
9295 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 7]  
9295 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Opening JDBC Connection 
9295 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 2]  
9296 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Checked out connection 19739814 from pool. 
9296 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Setting autocommit to false on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9296 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 0]  
9296 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 8]  
9296 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 4]  
9298 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - beginCommit...... txid: 1 
9298 [Thread-21] DEBUG c.a.storm.db.LocationDBStateUpdater - ^%*&^$&^#&%$@#^%$#^%$*&^%(&^%(&^%(&%(&%&  > [1, 29, 2011-01-27 00:00:00.0, 6, 0]  
9299 [Thread-21] DEBUG c.a.storm.db.LocationDBStateUpdater - ^%*&^$&^#&%$@#^%$#^%$*&^%(&^%(&^%(&%(&%&  > [1, 29, 2011-01-27 00:00:00.0, 0, 7]  
9299 [Thread-21] DEBUG c.a.storm.db.LocationDBStateUpdater - ^%*&^$&^#&%$@#^%$#^%$*&^%(&^%(&^%(&%(&%&  > [1, 29, 2011-01-27 00:00:00.0, 0, 2]  
9299 [Thread-21] DEBUG c.a.storm.db.LocationDBStateUpdater - ^%*&^$&^#&%$@#^%$#^%$*&^%(&^%(&^%(&%(&%&  > [1, 29, 2011-01-28 00:00:00.0, 23, 0]  
9299 [Thread-21] DEBUG c.a.storm.db.LocationDBStateUpdater - ^%*&^$&^#&%$@#^%$#^%$*&^%(&^%(&^%(&%(&%&  > [1, 29, 2011-01-28 00:00:00.0, 0, 8]  
9299 [Thread-21] DEBUG c.a.storm.db.LocationDBStateUpdater - ^%*&^$&^#&%$@#^%$#^%$*&^%(&^%(&^%(&%(&%&  > [1, 29, 2011-01-28 00:00:00.0, 0, 4]  
9299 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - save ......... START 
9300 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - #@#@@@@@@@@@@@@@@@@@@@@@@@@@ TEST TUPLE: [1, 29, 2011-01-27 00:00:00.0, 6, 0] 
9300 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - #@#@@@@@@@@@@@@@@@@@@@@@@@@@ TEST TUPLE: [1, 29, 2011-01-27 00:00:00.0, 0, 7] 
9300 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - #@#@@@@@@@@@@@@@@@@@@@@@@@@@ TEST TUPLE: [1, 29, 2011-01-27 00:00:00.0, 0, 2] 
9300 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - #@#@@@@@@@@@@@@@@@@@@@@@@@@@ TEST TUPLE: [1, 29, 2011-01-28 00:00:00.0, 23, 0] 
9300 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - #@#@@@@@@@@@@@@@@@@@@@@@@@@@ TEST TUPLE: [1, 29, 2011-01-28 00:00:00.0, 0, 8] 
9300 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - #@#@@@@@@@@@@@@@@@@@@@@@@@@@ TEST TUPLE: [1, 29, 2011-01-28 00:00:00.0, 0, 4] 
9301 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - save ......... END 
9301 [Thread-21] INFO  c.archermobile.storm.db.LocationDB - commit...... txid: 1 
9447 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ooo Using Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9447 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==>  Preparing: SELECT MAX(id) as max_id FROM msisdn_status  
9447 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==> Parameters:  
9599 [Thread-23] INFO  c.archermobile.storm.StatusManager - Max id: 2902288 
9599 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Resetting autocommit to true on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9750 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Closing JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9750 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Returned connection 19739814 to pool. 
9751 [Thread-23] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - emitPartitionBatch, startId: 50, endId: 100, max id: 2902288, TransactionAttempt: 2:0, partition: 0  
9751 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Opening JDBC Connection 
9751 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Checked out connection 19739814 from pool. 
9751 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Setting autocommit to false on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9902 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ooo Using Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
9902 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ==>  Preparing: SELECT sum(f.messages_submitted) as messages_submitted, sum(f.messages_failed) as messages_failed, fb.booking_id,f.classification_id,f.submission_account_id, f.timestamp FROM ( SELECT count(IF(msisdn_status.status >= 0, 1, NULL)) as messages_submitted, count(IF(msisdn_status.status < 0, 1, NULL)) as messages_failed, msisdn_status.flight_id, date(msisdn_status.timestamp) as timestamp, msisdn_status.classification_id, msisdn_status.submission_account_id FROM msisdn_status WHERE ( msisdn_status.id > ? AND msisdn_status.id <= ? AND msisdn_status.timestamp IS NOT NULL) GROUP BY flight_id, timestamp, classification_id, submission_account_id ) f LEFT JOIN jupiter.flight_reference fr on f.flight_id = fr.flight_id LEFT JOIN jupiterbs.flightboard fb on fb.flight_reference = fr.flight_reference GROUP BY f.timestamp, fb.booking_id, f.classification_id, f.submission_account_id;  
9903 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ==> Parameters: 50(Long), 100(Long) 
10087 [Thread-23] INFO  c.archermobile.storm.StatusManager -  Submitted List [MsisdnStatus{29, 2011-01-28 00:00:00.0, -1, -1, 35, 0}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 1, -1, 0, 6}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 2, -1, 0, 5}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 11, -1, 4, 0}] 
10087 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Resetting autocommit to true on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10238 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Closing JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10238 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Returned connection 19739814 to pool. 
10239 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Opening JDBC Connection 
10239 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 0]  
10239 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Checked out connection 19739814 from pool. 
10239 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 6]  
10239 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 5]  
10239 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 0]  
10239 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Setting autocommit to false on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10393 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ooo Using Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10393 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==>  Preparing: SELECT MAX(id) as max_id FROM msisdn_status  
10393 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==> Parameters:  
10546 [Thread-23] INFO  c.archermobile.storm.StatusManager - Max id: 2902288 
10546 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Resetting autocommit to true on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10698 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Closing JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10699 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Returned connection 19739814 to pool. 
10699 [Thread-23] INFO  c.a.s.MessageSubmittedOpaqueTransSpout - emitPartitionBatch, startId: 100, endId: 150, max id: 2902288, TransactionAttempt: 3:0, partition: 0  
10699 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Opening JDBC Connection 
10699 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Checked out connection 19739814 from pool. 
10699 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Setting autocommit to false on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10853 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ooo Using Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
10854 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ==>  Preparing: SELECT sum(f.messages_submitted) as messages_submitted, sum(f.messages_failed) as messages_failed, fb.booking_id,f.classification_id,f.submission_account_id, f.timestamp FROM ( SELECT count(IF(msisdn_status.status >= 0, 1, NULL)) as messages_submitted, count(IF(msisdn_status.status < 0, 1, NULL)) as messages_failed, msisdn_status.flight_id, date(msisdn_status.timestamp) as timestamp, msisdn_status.classification_id, msisdn_status.submission_account_id FROM msisdn_status WHERE ( msisdn_status.id > ? AND msisdn_status.id <= ? AND msisdn_status.timestamp IS NOT NULL) GROUP BY flight_id, timestamp, classification_id, submission_account_id ) f LEFT JOIN jupiter.flight_reference fr on f.flight_id = fr.flight_id LEFT JOIN jupiterbs.flightboard fb on fb.flight_reference = fr.flight_reference GROUP BY f.timestamp, fb.booking_id, f.classification_id, f.submission_account_id;  
10854 [Thread-23] DEBUG c.a.s.d.J.getSubmittedStatistics - ==> Parameters: 100(Long), 150(Long) 
11036 [Thread-23] INFO  c.archermobile.storm.StatusManager -  Submitted List [MsisdnStatus{29, 2011-01-28 00:00:00.0, -1, -1, 39, 0}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 1, -1, 0, 4}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 2, -1, 0, 1}, MsisdnStatus{29, 2011-01-28 00:00:00.0, 11, -1, 4, 2}] 
11036 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Resetting autocommit to true on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
11188 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Closing JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
11189 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Returned connection 19739814 to pool. 
11189 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Opening JDBC Connection 
11189 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Checked out connection 19739814 from pool. 
11189 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Setting autocommit to false on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
11190 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 0]  
11190 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 4]  
11190 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 1]  
11190 [Thread-21] INFO  c.a.storm.utils.PrintFilter - success submitted - start:  > [29, 2]  
11340 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ooo Using Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
11341 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==>  Preparing: SELECT MAX(id) as max_id FROM msisdn_status  
11341 [Thread-23] DEBUG c.a.s.d.J.getMaxSubmittedId - ==> Parameters:  
11494 [Thread-23] INFO  c.archermobile.storm.StatusManager - Max id: 2902288 
11494 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Resetting autocommit to true on JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
11646 [Thread-23] DEBUG o.a.i.t.jdbc.JdbcTransaction - Closing JDBC Connection [com.mysql.jdbc.JDBC4Connection@12d34a6] 
11646 [Thread-23] DEBUG o.a.i.d.pooled.PooledDataSource - Returned connection 19739814 to pool. 
..............

在日志的开头，您可以看到其他信息，如：

0    [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2000] WARN  org.apache.zookeeper.server.NIOServerCnxn  - EndOfStreamException: Unable to read additional data from client sessionid 0x13fcd101957000d, likely client has closed socket

我的问题和这个警告有关吗？我做错了什么？也许有另一种方法可以将结果存储在数据库中？

JavaDatabasestateapache-storm

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

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

    最新问答

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