Hmmm, I admit I'm a bit puzzled by this one... Could it be that the concept of equality at Storm level would not be aligned with equality at Microsoft SQL server level?
For instance, am I right guessing that "sendDateTime" is a java.util.Date that gets converted into a java.sql.Timestamp within the MssqlMapState by the following line: java.sql.Timestamp newParam = new java.sql.Timestamp(((Date) param).getTime ()); ? I'm not very familiar with java.sql.Timestamp, could it be that it gets stored with less precision than java.util.Date, so that the Storm groupBy would create as many groups as there are different milliseconds for sendDateTime but they would all point to the same timestamp, thus creating collisions? Maybe try to have HourTimeBucket output String instead of Date and see if it solves the issue. If not, try to remove all but one field in your groupBy statement, then add the others one by one until the deadlocks start re-appearing, then investigate how the latest added field is different in Java and in MSSQL. I hope this helps! Svend http://svendvanderveken.wordpress.com/ On Fri, Feb 14, 2014 at 11:48 AM, Danijel Schiavuzzi <[email protected] > wrote: > 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 >
