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.