Hi Svend, I use the this TridentState implementation: https://github.com/dschiavu/trident-mssql, which persists the state to a Microsoft SQL Server database table. It's a fork of https://github.com/wilbinsc/storm-mysql.
My topology describes a relatively common scenario, it sources some timed events from Kafka, aggregates them into discrete time buckets of one hour size, groups the records on that timebucket and some other keys, and uses persistentAggregate() to aggregate the count and persist it into the database. The spout is a transactional Trident Kafka spout (this one: https://github.com/wurstmeister/storm-kafka-0.8-plus) which pulls messages from a partitioned Kafka topic with two partitions, and the TridentState implementation is a transactional one too, matching the spout this regard. The topology definition is basically the following: // Kafka configuration BrokerHosts brokerHosts = new ZkHosts(kafkaZookeeper); TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(brokerHosts, kafkaTopic, "storm"); kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme()); TridentState tridentState = topology.newStream("KafkaMessages", new TransactionalTridentKafkaSpout(kafkaConfig)).name("KafkaMessages stream") .parallelismHint(2) .each(new Fields(), new ThroughputLoggingFilter()).name("ThroughputLoggingFilter after spout") .parallelismHint(1) .each(new Fields("str"), new JsonMessageParser(), new Fields("origSendDateTime", "channelId", "networkId", "countryId", "statusId", "requestTypeId", "count")) .parallelismHint(2) .each(new Fields("origSendDateTime"), new HourTimeBucket(), new Fields("sendDateTime")).name("HourTimeBucket function") .parallelismHint(2) .project(new Fields("sendDateTime", "channelId", "networkId", "countryId", "statusId", "requestTypeId", "count")) .groupBy(new Fields("sendDateTime", "channelId", "networkId", "countryId", "statusId", "requestTypeId")).name("group by") .persistentAggregate(MssqlState.newFactory(), new Fields("count"), new Sum(), new Fields("countTotal")) .parallelismHint(2); The SQL TridentState implements multiGet() by doing a batch UPSERT (INSERT or UPDATE) using MS SQL's MERGE statement. It works fine with paralellismHint(1) of course, but increasing the paralellism leads to database deadlocks -- probably caused by the two state instances updating the same records (keys) in the table, which, if I correctly understand the workings of the groupBy() statement, shouldn't occur. Any help understanding this is appreciated, Danijel On Fri, Feb 14, 2014 at 8:58 AM, Svend Vanderveken < [email protected]> wrote: > Hi Danijel, > > I think your description of the expected behavior of Trident primitives is > correct. I'm interested in understanding the issue you describe, could you > share some code so we can have a look? > > Svend > > > On Thu, Feb 13, 2014 at 9:20 PM, Danijel Schiavuzzi <[email protected]>wrote: > >> Hi, >> >> I'm experimenting a little with a TridentState implementation which >> persists state in a RDBMS (SQL) database, and it's working fine using >> paralellismHint(1). >> >> However, increasing the TridentState paralellism leads to database >> deadlocks, which is unusual since I thought that, by using groupBy() before >> persistentAggregate(), I was making sure that every TridentState instance >> receives a stream of distinct values at any given point in time -- so that >> every instance updates those distinct keys in the database (i.e. distinct >> table rows), thus avoiding any database concurrency issues. Apparently, >> this does not work as intended. >> >> Before I continue my research any further, I'd welcome any info or hint >> on avoiding these deadlocks. Should I improve my TridentState >> implementation somewhere? I see that the State is passed the partitionIndex >> and numPartition parameters on initialization, which I currently don't use >> -- and which I suppose are the key to implementing paralellism right? >> >> Also, could you recommend an example TridentState implementation which >> got the paralellism right? >> >> Any help is appreciated, >> >> -- >> Danijel Schiavuzzi >> > > -- Danijel Schiavuzzi E: [email protected] W: www.schiavuzzi.com T: +385989035562 Skype: danijels7
