Boil down events to time intervals

Question

Scenario: I have a service that logs events like in this CSV example:

#TimeStamp, Name, ColorOfPullover
TimeStamp01, Peter, Green
TimeStamp02, Bob, Blue
TimeStamp03, Peter, Green
TimeStamp04, Peter, Red
TimeStamp05, Peter, Green

Events that e.g. Peter wears Green will occur very often in a row.

I have two goals:

Keep the data as small as possible
Keep all the relevant data

Relevant means: I need to know, in which time spans a person was wearing what color. E.g:

#StartTime, EndTime, Name, ColorOfPullover
TimeStamp01, TimeStamp03, Peter, Green
TimeStamp02, TimeStamp02, Bob, Blue
TimeStamp03, TimeStamp03, Peter, Green
TimeStamp04, TimeStamp04, Peter, Red
TimeStamp05, TimeStamp05, Peter, Green

In this format, I can answer questions like: Which color was Peter wearing at time TimeStamp02? (I can safely assume that each person wears the same color in between two logged events for the same color.)

Main question: Can I use an already existing technology to accomplish that? I.e. I can supply it with a continuous stream of events and it extracts and stores the relevant data?

To be precise, I need to implement an algorithm like this (pseudocode). The OnNewEvent method is called for each line of the CSV example. Where parameter event already contains the data from the line as member variables.

def OnNewEvent(even)
    entry = Database.getLatestEntryFor(event.personName)
    if (entry.pulloverColor == event.pulloverColor)
        entry.setIntervalEndDate(event.date)
        Database.store(entry)
    else
        newEntry = new Entry
        newEntry.setIntervalStartDate(event.date)
        newEntry.setIntervalEndDate(event.date)
        newEntry.setPulloverColor(event.pulloverColor))
        newEntry.setName(event.personName)
        Database.createNewEntry(newEntry)
    end
end

It should be possible to do it with logstash, but the problem is that you'll have to do an elasticsearch request for each line to retrieve the latest entry, which will make the process very slow. That's why I don't think that logstash is the right tool for this. — baudsp, Sep 21 '17 at 16:16
What are you data volumes and how quickly do you need to react when a new event occurs? Is it ok if some events are lost? — Oleg Kuralenko, Sep 27 '17 at 19:07
Reaction to events may be slow. E.g. 1 day delay is acceptable. So, a cron job one a day could be an option. Events may not be lost, that is mission critical. — fex, Sep 28 '17 at 11:05

KrazyGautam · Answer 1 · 2017-09-26T10:04:09.643

This is typical scenario of any streaming architecture.  

There are multiple existing technologies which work in tandem  to get what you want. 


1.  NoSql Database (Hbase, Aerospike, Cassandra)
2.  streaming jobs Like Spark streaming(micro batch), Storm 
3.  Run mapreduce in micro batch to insert into NoSql Database.
4.  Kafka Distriuted queue

The end to end flow. 

Data -> streaming framework -> NoSql Database. 
OR 
Data -> Kafka -> streaming framework -> NoSql Database. 


IN NoSql database there are two ways to model your data. 
1. Key by "Name" and for every event for that given key, insert into Database.
   While fetching u get back all events corresponding to that key. 

2. Key by "name", every time a event for key is there, do a UPSERT into a existing blob(Object saved as binary), Inside the blob you maintain the time range and color seen.  

Code sample to read and write to Hbase and Aerospike

Hbase: http://bytepadding.com/hbase/

Aerospike: http://bytepadding.com/aerospike/

Sorry mate , hackers were having fun, just got the site fixed. feel free to go through the examples. and le me know if you need more clarification — KrazyGautam, Sep 28 '17 at 18:52

score 0 · Answer 2 · answered Sep 27 '17 at 17:29

One way to do it is to use HiveMQ. HiveMQ is a MQTT based message queue technology. Nice thing about it is you can write custom plugins to process incoming message. To get the latest entry of a event for a person, a hash table in HiveMQ plugin would work fine. If number of different persons is very large, I would consider using a cache like Redis to cache the latest event for each person.

You service publishes events to HiveMQ. HiveMQ plugin processes incoming event, and make updates to your database.

HiveMQ Plugin

Redis

Boil down events to time intervals

2 Answers2