Hi Svend, Exactly, the groupBy() fields are all part of a unique key in the database table.
Well, I guess some deadlocks are unavoidable, especially in such a high concurrency environment. Of course, my strategy is to eliminate deadlocks as much as possible, by optimizing the database queries etc. I did manage to minimize them so now I rely on Trident's transactional topology batch replay functionality to replay the failed batches. In any case, my throughput is now acceptable (for an SQL database) -- I guess I'll consider a NoSQL solution such as Cassandra should my throughput needs increase. Thanks for you help, Best regards, Danijel On Wed, Feb 19, 2014 at 9:04 AM, Svend Vanderveken < [email protected]> wrote: > Hi Danijel, > > I don't exactly know how Storm internally organises the various calls to > multiput, as I see it they are pretty much independent from each other and > executed by various tasks (i.e. thread) so overlapping in time, although > they should not overlap in terms of grouped value. > > Another idea concerning your topology (might sound silly, but that's all I > can think of now): do you confirm all the field defined in the groupBy are > configured as primary key in MSSQL? I see MSSqlState is meant to receive a > Config object specifying the key column => is it correct to assume this > requires to align groupby with MSSQlConfig with MSSQL schema definition? If > so, do you confirm it's the case for this topology? > > > Cheers, > > S > > > > > > > On Mon, Feb 17, 2014 at 6:45 PM, Danijel Schiavuzzi < > [email protected]> wrote: > >> Hi Svend, >> >> Thank you for your reply. >> >> Unfortunately, the problem doesn't seem to be in the handling of dates at >> the Java and database levels. I've confirmed this by eliminating all >> java.sql.Date and java.util.Date types from my topology entirely, instead >> rappresenting the timestamps as integers (milliseconds from epoch). >> >> My topology now works only with primitive types (integers) but database >> deadlocks still occur when paralellismHint is > 1 on the MS SQL >> TridentState. >> >> Am I correct in assuming that Storm invokes each TridentState's >> instance's multiPut() method in paralell, waiting for all instances' >> multiPut() methods to finish before actually beginning with the next batch? >> Or do the multiPut()s from the various instances get "interleaved"? >> >> I don't see how database deadlock could occur if I have the two >> TridentState instances updating distinct keys in the same time window. >> >> Of course, I'm assuming (1) the keys being updated by multiple >> TridentState instances are dinstinct, and (2) the multiPut()s for the same >> batch ID are being executed in the same time window by Storm. >> >> Thanks, >> >> Danijel >> >> >> On Fri, Feb 14, 2014 at 3:41 PM, Svend Vanderveken < >> [email protected]> wrote: >> >>> >>> >>> 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 >>>> >>> >>> >> >> >> -- >> Danijel Schiavuzzi >> >> E: [email protected] >> W: www.schiavuzzi.com >> T: +385989035562 >> Skype: danijels7 >> > > -- Danijel Schiavuzzi
