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

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.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 {

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 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()
}
}
}else if(lsUserInfo.size()==0){
}
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)

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.

• how is it possible that the aggregated data doesn't fit in memory? It's ~20 bytes per user - you really have a userbase of billions? – Oh My Goodness Apr 14 '19 at 12:37
• This is to bring out memory efficient solution and critical thinking among programmers in one of our internal org forum. – Wiki_91 Apr 14 '19 at 14:25
• is the problem real or imaginary? – Oh My Goodness Apr 14 '19 at 14:46
• Imaginery.. We handle such huge volume in distributed Hadoop cluster with spark. But this challenge is to avoid and handle the same solution in single machine. – Wiki_91 Apr 14 '19 at 15:45
• you've taken a real problem and applied made-up constraints, like a programming puzzle would have, and got the worst of both worlds. The arbitrary One True Solution character of a puzzle is combined with the vagueness, length and tedium of a real problem. I suggest to remove a bunch of detail to create a short puzzle, or drop the fake restrictions and add real context like "actual size of input" and "actual available memory" to describe an authentic engineering problem. – Oh My Goodness Apr 14 '19 at 15:52

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).

• I'd suggest adding the maxsplit parameter to split to deal with extra fields in the input lines. – Austin Hastings Apr 17 '19 at 0:05
• @AustinHastings That would just change it from being identified as a malformed line to an unknown event? And I would still need the try...except for lines with less fields (like empty lines). – Graipher Apr 17 '19 at 6:37

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.) 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.