Computing average duration per user from a large CSV file that doesn't fit in memory

Question

We have a large log file which stores user interactions with an application. The entries in the log file follow the following schema: {userId, timestamp, actionType} where actionType is one of two possible values: [open, close]

Constraints:

1. The log file is too big to fit in memory on one machine. Also assume that the aggregated data doesn’t fit into memory.
2. Code has to be able to run on a single machine.
3. Should not use an out-of-the box implementation of mapreduce or 3rd party database; don’t assume we have a Hadoop or Spark or other distributed computing framework.
4. There can be multiple entries of each actionType for each user, and there might be missing entries in the log file. So a user might be missing a close record between two open records or vice versa.
5. Timestamps will come in strictly ascending order.

For this problem, we need to implement a class/classes that computes the average time spent by each user between open and close. Keep in mind that there are missing entries for some users, so we will have to make a choice about how to handle these entries when making our calculations. Code should follow a consistent policy with regards to how we make that choice.

The desired output for the solution should be [{userId, timeSpent},….] for all the users in the log file.

Sample log file (comma-separated, text file)

1,1435456566,open 
2,1435457643,open 
3,1435458912,open 
1,1435459567,close 
4,1435460345,open 
1,1435461234,open 
2,1435462567,close 
1,1435463456,open 
3,1435464398,close 
4,1435465122,close 
1,1435466775,close

Approach

Here is the code I've written in Python and Scala, which seems to be not efficient and up to the expectations of the scenario given. I'd like feedback on how I could optimise this code as per the given scenario.

Scala implementation

import java.io.FileInputStream
import java.util.{Scanner, Map, LinkedList}
import java.lang.Long
import scala.collection.mutable

object UserMetrics extends App {
  if (args.length == 0) {
    println("Please provide input data file name for processing")
  } 
  val userMetrics = new UserMetrics()
  userMetrics.readInputFile(args(0),if (args.length == 1) 600000 else args(1).toInt)
}

case class UserInfo(userId: Integer, prevTimeStamp: Long, prevStatus: String, timeSpent: Long, occurence: Integer)

class UserMetrics {

  val usermap = mutable.Map[Integer, LinkedList[UserInfo]]()

  def readInputFile(stArr:String, timeOut: Int) {
    var inputStream: FileInputStream = null
    var sc: Scanner = null
    try {
      inputStream = new FileInputStream(stArr);
      sc = new Scanner(inputStream, "UTF-8");
      while (sc.hasNextLine()) {
        val line: String = sc.nextLine();
        processInput(line, timeOut)
      }

      for ((key: Integer, userLs: LinkedList[UserInfo]) <- usermap) {
        val userInfo:UserInfo = userLs.get(0)
        val timespent = if (userInfo.occurence>0) userInfo.timeSpent/userInfo.occurence else 0
        println("{" + key +","+timespent + "}")
      }

      if (sc.ioException() != null) {
        throw sc.ioException();
      }
    } finally {
      if (inputStream != null) {
        inputStream.close();
      }
      if (sc != null) {
        sc.close();
      }
    }
  }

  def processInput(line: String, timeOut: Int) {
    val strSp = line.split(",")

    val userId: Integer = Integer.parseInt(strSp(0))
    val curTimeStamp = Long.parseLong(strSp(1))
    val status = strSp(2)
    val uInfo: UserInfo = UserInfo(userId, curTimeStamp, status, 0, 0)
    val emptyUserInfo: LinkedList[UserInfo] = new LinkedList[UserInfo]()

    val lsUserInfo: LinkedList[UserInfo] = usermap.getOrElse(userId, emptyUserInfo)

    if (lsUserInfo != null && lsUserInfo.size() > 0) {
      val lastUserInfo: UserInfo = lsUserInfo.get(lsUserInfo.size() - 1)
      val prevTimeStamp: Long = lastUserInfo.prevTimeStamp
      val prevStatus: String = lastUserInfo.prevStatus

      if (prevStatus.equals("open")) {
        if (status.equals(lastUserInfo.prevStatus)) {
           val timeSelector = if ((curTimeStamp - prevTimeStamp) > timeOut) timeOut else curTimeStamp - prevTimeStamp
           val timeDiff = lastUserInfo.timeSpent + timeSelector
          lsUserInfo.remove()
          lsUserInfo.add(UserInfo(userId, curTimeStamp, status, timeDiff, lastUserInfo.occurence + 1))
        } else if(!status.equals(lastUserInfo.prevStatus)){
          val timeDiff = lastUserInfo.timeSpent + curTimeStamp - prevTimeStamp
          lsUserInfo.remove()
          lsUserInfo.add(UserInfo(userId, curTimeStamp, status, timeDiff, lastUserInfo.occurence + 1))
        }
      } else if(prevStatus.equals("close")) {
        if (status.equals(lastUserInfo.prevStatus)) {
          lsUserInfo.remove()
          val timeSelector = if ((curTimeStamp - prevTimeStamp) > timeOut) timeOut else curTimeStamp - prevTimeStamp
          lsUserInfo.add(UserInfo(userId, curTimeStamp, status, lastUserInfo.timeSpent + timeSelector, lastUserInfo.occurence+1))
        }else if(!status.equals(lastUserInfo.prevStatus))
          {     
          lsUserInfo.remove()
          lsUserInfo.add(UserInfo(userId, curTimeStamp, status, lastUserInfo.timeSpent, lastUserInfo.occurence))
        }
      }
    }else if(lsUserInfo.size()==0){
      lsUserInfo.add(uInfo)
    }
    usermap.put(userId, lsUserInfo)
  }

}

Python Implementation

import sys

def fileBlockStream(fp, number_of_blocks, block):
    #A generator that splits a file into blocks and iterates over the lines of one of the blocks.

    assert 0 <= block and block < number_of_blocks #Assertions to validate number of blocks given
    assert 0 < number_of_blocks

    fp.seek(0,2) #seek to end of file to compute block size
    file_size = fp.tell() 

    ini = file_size * block / number_of_blocks #compute start & end point of file block
    end = file_size * (1 + block) / number_of_blocks

    if ini <= 0:
        fp.seek(0)
    else:
        fp.seek(ini-1)
        fp.readline()

    while fp.tell() < end:
        yield fp.readline() #iterate over lines of the particular chunk or block

def computeResultDS(chunk,avgTimeSpentDict,defaultTimeOut):
    countPos,totTmPos,openTmPos,closeTmPos,nextEventPos = 0,1,2,3,4
    for rows in chunk.splitlines():
        if len(rows.split(",")) != 3:
            continue
        userKeyID = rows.split(",")[0]
        try:
            curTimeStamp = int(rows.split(",")[1])
        except ValueError:
            print("Invalid Timestamp for ID:" + str(userKeyID))
            continue
        curEvent = rows.split(",")[2]
        if userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==1 and curEvent == "close": 
        #Check if already existing userID with expected Close event 0 - Open; 1 - Close
        #Array value within dictionary stores [No. of pair events, total time spent (Close tm-Open tm), Last Open Tm, Last Close Tm, Next expected Event]
            curTotalTime = curTimeStamp - avgTimeSpentDict[userKeyID][openTmPos]
            totalTime = curTotalTime + avgTimeSpentDict[userKeyID][totTmPos]
            eventCount = avgTimeSpentDict[userKeyID][countPos] + 1
            avgTimeSpentDict[userKeyID][countPos] = eventCount
            avgTimeSpentDict[userKeyID][totTmPos] = totalTime
            avgTimeSpentDict[userKeyID][closeTmPos] = curTimeStamp
            avgTimeSpentDict[userKeyID][nextEventPos] = 0 #Change next expected event to Open

        elif userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==0 and curEvent == "open":
            avgTimeSpentDict[userKeyID][openTmPos] = curTimeStamp
            avgTimeSpentDict[userKeyID][nextEventPos] = 1 #Change next expected event to Close

        elif userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==1 and curEvent == "open":
            curTotalTime,closeTime = missingHandler(defaultTimeOut,avgTimeSpentDict[userKeyID][openTmPos],curTimeStamp)
            totalTime = curTotalTime + avgTimeSpentDict[userKeyID][totTmPos]
            avgTimeSpentDict[userKeyID][totTmPos]=totalTime
            avgTimeSpentDict[userKeyID][closeTmPos]=closeTime
            avgTimeSpentDict[userKeyID][openTmPos]=curTimeStamp
            eventCount = avgTimeSpentDict[userKeyID][countPos] + 1
            avgTimeSpentDict[userKeyID][countPos] = eventCount          

        elif userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==0 and curEvent == "close": 
            curTotalTime,openTime = missingHandler(defaultTimeOut,avgTimeSpentDict[userKeyID][closeTmPos],curTimeStamp)
            totalTime = curTotalTime + avgTimeSpentDict[userKeyID][totTmPos]
            avgTimeSpentDict[userKeyID][totTmPos]=totalTime
            avgTimeSpentDict[userKeyID][openTmPos]=openTime
            eventCount = avgTimeSpentDict[userKeyID][countPos] + 1
            avgTimeSpentDict[userKeyID][countPos] = eventCount

        elif curEvent == "open":
            #Initialize userid with Open event
            avgTimeSpentDict[userKeyID] = [0,0,curTimeStamp,0,1]

        elif curEvent == "close":
            #Initialize userid with missing handler function since there is no Open event for this User
            totaltime,OpenTime = missingHandler(defaultTimeOut,0,curTimeStamp)
            avgTimeSpentDict[userKeyID] = [1,totaltime,OpenTime,curTimeStamp,0]

def missingHandler(defaultTimeOut,curTimeVal,lastTimeVal):
    if lastTimeVal - curTimeVal > defaultTimeOut:
        return defaultTimeOut,curTimeVal
    else:
        return lastTimeVal - curTimeVal,curTimeVal

def computeAvg(avgTimeSpentDict,defaultTimeOut):
    resDict = {}
    for k,v in avgTimeSpentDict.iteritems():
        if v[0] == 0:
            resDict[k] = 0
        else:
            resDict[k] = v[1]/v[0]
    return resDict

if __name__ == "__main__":
    avgTimeSpentDict = {}
    if len(sys.argv) < 2:
        print("Please provide input data file name for processing")
        sys.exit(1)

    fileObj = open(sys.argv[1])
    number_of_chunks = 4 if len(sys.argv) < 3 else int(sys.argv[2])
    defaultTimeOut = 60000 if len(sys.argv) < 4 else int(sys.argv[3])
    for chunk_number in range(number_of_chunks):
        for chunk in fileBlockStream(fileObj, number_of_chunks, chunk_number):
            computeResultDS(chunk, avgTimeSpentDict, defaultTimeOut)
    print (computeAvg(avgTimeSpentDict,defaultTimeOut))
    avgTimeSpentDict.clear() #Nullify dictionary 
    fileObj.close #Close the file object

Both programs give the desired output, but efficiency is what matters for this particular scenario. Let me know if you have anything better or any suggestions on the existing implementation.

Answer 1

This is a comment on your Python solution (I don't know anything about Scala).

You don't need to iterate over chunks of your file unless you want to do parallel processing. However, since there might be a close event in a different block from an opening event, this process is not so easy to parallelize (you would have to keep track of dangling users in both directions, which you don't do as far as I can tell).

Also, the restriction that the aggregate does not fit into memory is...unrealistic IMO. You would have to have more users than there are people in the world. Anyways, your code does not respect this constraint either, since avgTimeSpentDict contains all users and will therefore not fit into memory. So I'm going to ignore this part.

Instead, just iterate over the file normally, with a for loop. This does not read the whole file into memory. Update a running mean with the new value whenever you find a matching event for each user.

At the same time keep a dictionary of users that are open to look out for a matching close event. If you have a close event without an open, it is a broken one and we can ignore it because you said it is guaranteed that the times are sorted (and time travel has not been invented yet, AFAIK). Or do something else with it. Same goes for an open event after a previous open, without any intervening close. Here I just added a print in those cases.

import sys
from collections import defaultdict

def update_mean(count, mean, new_value):
    count += 1.  # float so it also works in Python 2
    mean += (new_value - mean) / count
    return count, mean

def average_timeout(file_name):    
    open_users = {}
    time_spent = defaultdict(lambda: (0., 0.))

    with open(file_name) as f:
        for line in f:
            print(line.strip())
            try:
                user_id, timestamp, event = line.strip().split(",")
            except ValueError:
                print(f"misformed line: {line!r}")
                continue

            if event == "open":
                if user_id in open_users:
                    print("open with prior open, missed a close")
                open_users[user_id] = int(timestamp)
            elif event == "close":
                if user_id not in open_users:
                    print("close without open")
                else:
                    diff = int(timestamp) - open_users.pop(user_id)
                    time_spent[user_id] = update_mean(*time_spent[user_id], diff)
                    print(f"close with prior open, time difference {diff}")
            else:
                print(f"Unknown event: {event}")

    print(f"{len(open_users)} users left without close event")
    return time_spent

if __name__ == "__main__":
    time_spent = average_timeout(sys.argv[1])
    for user, (_, mean) in time_spent.items():
        print(f"{user} average timeout: {mean}")

In production you will obviously want to either remove most of the prints or at least make them logging.debug calls.

This can still run out of memory if the average length between an open and a close event contains more open events by different users than there is memory. Or if all events are broken and lack a close.

Python has an official style-guide, PEP8, which programmers are encouraged to follow. It recommends using lower_case for functions and variables and putting a space after each comma in an argument list.

fileObj.close does not actually close the file if you don't call it, fileObj.close(). But even better is to use with which will take care of closing the file for you, even in the event of an exception occurring somewhere.

You should use Python 3. Python 2 will no longer be supported in less than a year.

You can use x in d to check if some value x is in a dictionary d. No need to do x in d.keys(). In Python 2 this distinction is even more important since x in d is \$\mathcal{O}(1)\$, while x in d.keys() is \$\mathcal{O}(n)\$ (since it is a list).

Answer 2

Graipher's answer already addresses some of the higher-level issues with your Python solution. Like him, I cannot speak to the Scala solution.

Instead, I'll focus on your core function and how to improve the speed of "C Python" code (that is, Python run under the standard python executable written in C).

The function:

def computeResultDS(chunk,avgTimeSpentDict,defaultTimeOut):
    countPos,totTmPos,openTmPos,closeTmPos,nextEventPos = 0,1,2,3,4
    for rows in chunk.splitlines():
        if len(rows.split(",")) != 3:
            continue
        userKeyID = rows.split(",")[0]
        try:
            curTimeStamp = int(rows.split(",")[1])
        except ValueError:
            print("Invalid Timestamp for ID:" + str(userKeyID))
            continue
        curEvent = rows.split(",")[2]
        if userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==1 and curEvent == "close": 
        #Check if already existing userID with expected Close event 0 - Open; 1 - Close
        #Array value within dictionary stores [No. of pair events, total time spent (Close tm-Open tm), Last Open Tm, Last Close Tm, Next expected Event]
            curTotalTime = curTimeStamp - avgTimeSpentDict[userKeyID][openTmPos]
            totalTime = curTotalTime + avgTimeSpentDict[userKeyID][totTmPos]
            eventCount = avgTimeSpentDict[userKeyID][countPos] + 1
            avgTimeSpentDict[userKeyID][countPos] = eventCount
            avgTimeSpentDict[userKeyID][totTmPos] = totalTime
            avgTimeSpentDict[userKeyID][closeTmPos] = curTimeStamp
            avgTimeSpentDict[userKeyID][nextEventPos] = 0 #Change next expected event to Open

        elif userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==0 and curEvent == "open":
            avgTimeSpentDict[userKeyID][openTmPos] = curTimeStamp
            avgTimeSpentDict[userKeyID][nextEventPos] = 1 #Change next expected event to Close

        elif userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==1 and curEvent == "open":
            curTotalTime,closeTime = missingHandler(defaultTimeOut,avgTimeSpentDict[userKeyID][openTmPos],curTimeStamp)
            totalTime = curTotalTime + avgTimeSpentDict[userKeyID][totTmPos]
            avgTimeSpentDict[userKeyID][totTmPos]=totalTime
            avgTimeSpentDict[userKeyID][closeTmPos]=closeTime
            avgTimeSpentDict[userKeyID][openTmPos]=curTimeStamp
            eventCount = avgTimeSpentDict[userKeyID][countPos] + 1
            avgTimeSpentDict[userKeyID][countPos] = eventCount          

        elif userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==0 and curEvent == "close": 
            curTotalTime,openTime = missingHandler(defaultTimeOut,avgTimeSpentDict[userKeyID][closeTmPos],curTimeStamp)
            totalTime = curTotalTime + avgTimeSpentDict[userKeyID][totTmPos]
            avgTimeSpentDict[userKeyID][totTmPos]=totalTime
            avgTimeSpentDict[userKeyID][openTmPos]=openTime
            eventCount = avgTimeSpentDict[userKeyID][countPos] + 1
            avgTimeSpentDict[userKeyID][countPos] = eventCount

        elif curEvent == "open":
            #Initialize userid with Open event
            avgTimeSpentDict[userKeyID] = [0,0,curTimeStamp,0,1]

        elif curEvent == "close":
            #Initialize userid with missing handler function since there is no Open event for this User
            totaltime,OpenTime = missingHandler(defaultTimeOut,0,curTimeStamp)
            avgTimeSpentDict[userKeyID] = [1,totaltime,OpenTime,curTimeStamp,0]

The Rules

The rules of performance-hunting in CPython are as follows:

0. Don't. CPython is slow, so the easiest way to improve performance is to not use CPython. Use Cython, pypy, or some other language.

1. Always go for the 'big O'. The best way to gain performance is by algorithmic improvements. If you can convert from \$O(n^2)\$ to \$O(n log n)\$ you will be better off than you would be trying to squeeze out incremental improvements.

2. Never do in Python that which you can do in 'C'. This is particularly relevant to your code because you have written I/O buffering code. For reading a text file. In Python. Don't do that. Let C handle it, and just use line-at-a-time mode for reading the file. (The exception to this would be rule #1. If you could convert from reading the entire file in C to bsearching the file in Python, it might be a win.)

2a. You can get a surprising amount of performance by switching to numpy or pandas. It's worth the learning curve if you really want to process large amounts of data in CPython.

3. It's all about the lookups. The Python spec defines a language that is basically impossible to optimize. Every name, including attribute and method names, has to be looked up every time. The compiler will only do the most rudimentary of expression folding, on the off chance that an invocation in one statement has caused the __iadd__ method to be replaced in a subsequent statement, so things like:

x += 1
x += 2

won't be folded. And things like:

avgTimeSpentDict[userKeyID][countPos] = eventCount
avgTimeSpentDict[userKeyID][totTmPos] = totalTime
avgTimeSpentDict[userKeyID][closeTmPos] = curTimeStamp
avgTimeSpentDict[userKeyID][nextEventPos] = 0 #Change next expected event to Open

Repeat the same lookups over, and over, and over, and over again. The solution to this is caching in local variables, which I'll demonstrate below.

4. Provide hard timings or get out! Some opcodes are faster than others. Some things pipeline better than others. Sometimes smaller code fits in the cache better than larger-but-theoretically-faster code.

If you are concerned about performance, the very first thing you must do is build a timing framework, or you're not really concerned about performance. In many cases, the current date/time before and after is enough- if your code doesn't take at least 1 second to run, you either don't have enough data in your test data set, or you don't really have a performance problem!

The changes

Memory usage

Graipher pointed out that your code doesn't follow the constraints you laid down. So let's talk about that first.

In order to try to honor your constraint about the aggregate user data not fitting in available memory, you should focus on building a pipeline. Let the first stage of your pipeline be something like this program. Unix provides the sort utility which can handle files that are larger than available memory, so if you focus on cleaning up the data and constructing a series of elapsed-time records with associated user id, you can then sort by the user id to get them adjacent, and the third stage of your pipeline could sum and average intervals for one user at a time. In Unix terms:

$ make-interval-records < infile | sort (by userid) | sum-and-average > outfile

Or, as with Graipher's answer, we could ignore this constraint. That's more relevant, I think.

Lookups

Here's a REPL session I just ran. I defined a function with the first few lines of your inner loop, and used the built-in module dis to dump the generated byte codes. As a rule, some bytecodes are slower than others, but more bytecodes is generally slower than less bytes codes.

h[1] >>> def f(chunk):
...     for rows in chunk:
...         if len(rows.split(",")) != 3:
...             continue
...         userKeyID = rows.split(",")[0]
...         try:
...             curTimeStamp = int(rows.split(",")[1])
...         except ValueError:
...             print("Invalid Timestamp for ID:" + str(userKeyID))
...             continue
...
h[1] >>> dis.dis(f)
  2           0 SETUP_LOOP             108 (to 110)
              2 LOAD_FAST                0 (chunk)
              4 GET_ITER
        >>    6 FOR_ITER               100 (to 108)
              8 STORE_FAST               1 (rows)

Ignore the opcodes above ^^ as they are just faked-up to model your loop.

  3          10 LOAD_GLOBAL              0 (len)
             12 LOAD_FAST                1 (rows)
             14 LOAD_METHOD              1 (split)
             16 LOAD_CONST               1 (',')
             18 CALL_METHOD              1
             20 CALL_FUNCTION            1
             22 LOAD_CONST               2 (3)
             24 COMPARE_OP               3 (!=)
             26 POP_JUMP_IF_FALSE       30

Notice that the first thing we do here is a global lookup for the name len. Yes, it's a built-in function. But Python allows for that function to be overridden at the module level, so it checks to see if it was overridden. Every. Single. Time.

On the other hand, the very next opcode is a LOAD_FAST of the rows variable. You might think that this was a faster operation than the LOAD_GLOBAL. I could not possibly comment.

When you're inside the innermost loop of your code and you want performance, you need to put every single thing in local variables. Especially the functions.

  4          28 JUMP_ABSOLUTE            6

  5     >>   30 LOAD_FAST                1 (rows)
             32 LOAD_METHOD              1 (split)
             34 LOAD_CONST               1 (',')
             36 CALL_METHOD              1
             38 LOAD_CONST               3 (0)
             40 BINARY_SUBSCR
             42 STORE_FAST               2 (userKeyID)

Notice the lookup and call to split? Remember that you did the same call to split on the previous line? Rule 3 strikes again! Lookups are repeated every single time. Function calls are not memoized unless you do it, or unless you're using a library whose documentation says "function calls are memoized."

If you compute a result, save it in a local variable until you're sure you're done with it.

  6          44 SETUP_EXCEPT            22 (to 68)

  7          46 LOAD_GLOBAL              2 (int)
             48 LOAD_FAST                1 (rows)
             50 LOAD_METHOD              1 (split)
             52 LOAD_CONST               1 (',')
             54 CALL_METHOD              1
             56 LOAD_CONST               4 (1)
             58 BINARY_SUBSCR
             60 CALL_FUNCTION            1
             62 STORE_FAST               3 (curTimeStamp)
             64 POP_BLOCK
             66 JUMP_ABSOLUTE            6

  8     >>   68 DUP_TOP
             70 LOAD_GLOBAL              3 (ValueError)
             72 COMPARE_OP              10 (exception match)
             74 POP_JUMP_IF_FALSE      104
             76 POP_TOP
             78 POP_TOP
             80 POP_TOP

The immediately preceding paragraph is "exception matching". It's a comparison of the exception received versus the exception expected. If you simply do except: ... there is no matching. It's probably worth it for situations where you absolutely know what exceptions are going to be raised.

  9          82 LOAD_GLOBAL              4 (print)
             84 LOAD_CONST               5 ('Invalid Timestamp for ID:')
             86 LOAD_GLOBAL              5 (str)
             88 LOAD_FAST                2 (userKeyID)
             90 CALL_FUNCTION            1
             92 BINARY_ADD
             94 CALL_FUNCTION            1
             96 POP_TOP

Lots of lookups and stuff to process this error. If errors are common, just append the relevant data to a list and post-process the errors.

 10          98 CONTINUE_LOOP            6
            100 POP_EXCEPT
            102 JUMP_ABSOLUTE            6
        >>  104 END_FINALLY
            106 JUMP_ABSOLUTE            6
        >>  108 POP_BLOCK
        >>  110 LOAD_CONST               0 (None)
            112 RETURN_VALUE

I'm not going to bother with the remainder of the function -- you can do it yourself with your own REPL. But I'll point out two glaring issues:

    if userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==1 and curEvent == "close": 

    elif userKeyID in avgTimeSpentDict.keys() and avgTimeSpentDict[userKeyID][nextEventPos]==0 and curEvent == "open":

First: never throw away data. In this case, you're checking if the user id is in the dictionary (in a pathologically bad way, see Graipher's answer) and then discarding that information. Also, you compare the event with "open" or "closed" in multiple places, and then throw it away. Stop doing that.

Next: Never repeat a lookup. Every time you write avgTimeSpentDict[userKeyID]... a kitten is murdered.

You should be striving to accomplish as much as you can with as few lookups as possible. Every significant function and data structure should be accessible through a single local variable. In many cases, bound methods can be loaded into variables as long as the underlying object is not changing. (Thus, the rows variable is not a candidate, but a list of errors' append method would be a candidate.)

Answer 3

This is a review of your Scala solution.

It appears to be written by someone who doesn't understand Scala (and Functional Programming) or just doesn't like it. Witness all the semicolons in the readInputFile() method.

There are a number of curious aspects to the code. I'll touch on just a few.

import java.util.{Scanner, Map, LinkedList}
import java.lang.Long

Why import Java's LinkedList and Long? Scala offers a native Long as well as a number of collection types with various performance characteristics.

val emptyUserInfo: LinkedList[UserInfo] = new LinkedList[UserInfo]()

val lsUserInfo: LinkedList[UserInfo] = usermap.getOrElse(userId, emptyUserInfo)

Why create an emptyUserInfo if you might use it? Why not create it only when you need it?

And there's this funny if/else if pattern that occurs several times.

if       (someCondition) ...
else if (!sameCondition) ...

Why test for a condition that is guaranteed to be true?

fault tolerance

In general, the code doesn't appear to be very resilient. It throws an error if the input file can't be found, or if any of the userId fields or the currTimeStamp fields can't be parsed. If the status field, on the other hand, is not as expected (maybe an unexpected space on the end) then the entire record is silently dropped without notice.

Scala style

Idiomatic Scala adheres to the principles of Functional Programming, in particular: limited mutability. It's also a good idea to use the type system to capture and manage the possibility of errors.

Here's an implementation of your basic task, redesigned with these principles in mind. This produces the same output as your code, as far as my limited testing can determine.

import util.Try

val inFormat = raw"\s*(\w+)\s*,\s*(\d+)\s*,\s*(open|close)\s*".r
case class Stats(timestamp :Long, accumulator :Long, sessionCount :Int)

val file = Try(io.Source.fromFile(fileName))

val data = file.map(_.getLines().flatMap{line =>
  Try{line match {case inFormat(a,b,c) => (a, if(c=="open") -b.toLong
                                              else           b.toLong)}
  }.fold(_ => {println(s"bad input: $line");None}, Some(_))
}.foldLeft(Map[String,Stats]().withDefaultValue(Stats(0,0,0))){
  case (m,(id,thisTS)) =>
    val Stats(prevTS, acc, cnt) = m(id)
    if (prevTS < 0 && thisTS < 0) {
      println("missing close")
      m + (id -> Stats(prevTS, acc, cnt+1))
    } else if (prevTS > 0 && thisTS > 0) {
      println("missing open")
      m + (id -> Stats(thisTS, acc - prevTS + thisTS, cnt+1))
    } else if (prevTS < 0 && thisTS > 0)  //open-->close
      m + (id -> Stats(thisTS, acc + prevTS + thisTS, cnt+1))
    else //close-->open
      m + (id -> Stats(thisTS, acc, cnt))
})

val result = file.map(_.close).flatMap(_ => data.map(_.foreach{
  case (id,Stats(_,total,count)) => println(s"{$id,${total/count}}")}))

The final result is either Failure(/*exception type here*/) or Success(()). (The Success is empty because println() returns Unit.) Also, faulty input is recognized and reported. There are no mutable variables or data structures.

It's not uncommon for Scala coders to forego cherished FP principles when efficiency and throughput becomes an issue, but it's usually a good idea to start with best practices and then deviate only when a good profiler indicates where it's necessary.