2

I have two async streams suppose- Trip : {tripId, date, city} Bill : {billId, tripId, date, amount}. I need to design a system to get real time aggregated view of following nature: City, TripCount, TotalAmount. Events in both streams can be out of sync or duplicate. But result needs to be accurate and realtime.

My Solution:

1.) Create two different DB tables: Trip and Bill (indexed on TripID and BillID). Read the messages from the streams and persist in these tables with the status column as pending in Bill table and Trip table. Then write a worker which will read from the bill table and look into the Trip table for the record containing the given tripID. If the record is found it will update the aggregated view in third table (City, TripCount, TotalAmount). Then we will change the state of the bill and trip record to processed. There will be one background job running on periodic basis which will remove all the records which will have the state as processed from both Bill and Trip table.

The problem i see with the above solution is the indexing that done on the TripID and BillID will become a bottle neck if i remove the records at very high frequency. Other from it do you guy see any other problem with this solution. I have read on internet that people are suggesting it as very famous anti-pattern cause here i am using DB as queue.

2.) Here is other solution: Take the data from the streams, persist it in the tables: Trip, Bill (for audit purpose of records and to avoid duplicates). For storing the Trip data temporarily take one distributed key-value pair very fast data structure. I am taking REDIS for this purpose. So, after writing trip data to DB, i will write the same data in the cache with tripid as key and record as value. Then i will put the bill data in the queue. Workers will be reading from the queue and will lookup into the cache for the tripid. If the tripid is present then the workers will read the data from cache and update the aggregated view and delete the tripid from the cache and also bill message from the queue. If the tripid is not found in the cache then then worker will again put the same message in the queue back.

To avoid duplicates, the insertion will fail if we will try to insert same tripid or billid in the tables. In case where the insertion will fail, i will not put the message in the queue and cache.

Experts please let me know about your thoughts on the above two solutions and please propose any better solution if you have any.

  • #2 is a dreadful solution. You'd be forcing your workers to continuously wade through the entire backlog of trips whether or not anything has changed, and new trips would be put at the back of the line. This could be fully developed in a RDBMS in an hour or two without have to resort to external workers, caches, queue systems or anything else like that. – Blrfl Aug 21 at 12:30
0

Use standard queue based system rather than scanning database for rows to process it will be efficient and can act as event store as well. Use Domain Events to decouple Bill, Trip related events from any other logic processing. Not sure why you are clearing the database once marked processed, it these are temporary tables then these can be replaced with Queue based event store.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.