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.