Tracking durations of login/logout sessions

edit

You could do this by writing session details to MySQL. Here is a sample pipeline I created:

I used a Dev Raw Data Source as the origin, configured with some sample JSON data:

{
  "type":"session_start",
  "user":12345,
  "session_id":23456,
  "timestamp":"2019-02-04 01:23:45.678"
}
{
  "type":"session_start",
  "user":34567,
  "session_id":45678,
  "timestamp":"2019-02-04 02:34:56.789"
}
{
  "type":"session_end",
  "user":12345,
  "session_id":23456,
  "timestamp":"2019-02-04 03:45:67.890"
}
{
  "type":"session_end",
  "user":34567,
  "session_id":45678,
  "timestamp":"2019-02-05 04:56:78.901"
}

The Field Type Converter converts the /timestamp string to a DATETIME:

The Stream Selector routes records according to whether their /type is session_start:

If '/type' is session start, then the record is written to a database via a JDBC Producer destination. I simply created a MySQL table to model the record structure:

CREATE TABLE session (
    session_id INT PRIMARY KEY,
    user INT,
    timestamp DATETIME
);

The default stream from the Stream Selector sends session_end records to a JDBC Lookup processor. A simple SQL query retrieves the relevant session start time, and I mapped /timestamp to '/session_start` so it wouldn't overwrite the session end time in the record:

An Expression Evaluator calculates the session duration in milliseconds. You could use a Field Renamer to rename '/timestamp' to '/session_end' to make things more tidy, but I just left it as is:

Finally, another JDBC Destination, with Default Operation set to DELETE, deletes the session from the database.

My sample pipeline uses a Local FS destination to write JSON output:

{"type":"session_end","user":12345,"session_id":23456,"timestamp":1549280767890,"session_start":1549272226000,"duration_milliseconds":8541890}
{"type":"session_end","user":34567,"session_id":45678,"timestamp":1549371438901,"session_start":1549276497000,"duration_milliseconds":94941901}

You would more likely write the session data to a Kafka topic or database table.