CareerCup

Presuming a protocol exists that can ask three questions to each server:

* the score of a single url
* the top 10
* the top n that satisfy score >= N

We program a two pass solution like so:

We denote the number of servers as S.

[First pass]
(1) Ask every server for its own top ten

(2) merge the results. For all URLs in the merged set calculate correct values by asking
all servers for their scores for each URL. Calculate a set of top ten from our sample.

(3) pick score of the now tenth URL as the threshold that we try to beat
in the second round. We denote the threshold as T.

[Second pass]
(4) Ask every server for all its top N that satisfy score >= T/S

(5) Merge these bigger samples again as in step (2)

(6) We now have the correct top ten with correct scores.

The constraints puzzle me a bit, especially the "using MapReduce directly, is not allowed" one. I would try to discuss what that means exactly in an interview.
I'll give another shot at the question:

Denote N as the number of computers in our network.

1) Pick a good string hash function. This function must take urls as input and produce hash values uniformly in the range [0, MAX_HASH_VALUE]

2) Divide the hash range [0, MAX_HASH_VALUE] into N intervals with equal length MAX_HASH_VALUE / N, and assign each interval to 1 computer in the network, e.g.
CPU_1) [0, length-1]
CPU_2) [length, 2*length-1]
CPU_3) [2*length, 3*length-1]
...
CPU_N) [(N-1)*length, MAX_HASH_VALUE]

3) Each computer computes the hash values of its list of urls and sends the url and its visited information to the computer responsible by the hash of that url.

4) Each computer received a list of information url->visits for the urls in its hash interval. Now it must combine the visits of the urls and produce its top 10.

5) Each computer sends its top 10 to a central computer. This central computer will receive 10*N urls and compute the overall top 10.

Due to our good hash function, we expect that each computer will receive roughly the same amount of urls to process in the 4th step. I think this approach is the best we can do to distribute the work among the cluster.

About the constraints, they're not very clear.
a) This is sending all urls over the network but not to a single computer. In order to produce an exact result, I think all approaches end up sending all urls over the network in the worst case. Again, we would need to discuss with the interviewer if this is ok (the "especially sending all of them to a central server" part).

b) This is similar to MapReduce. I think that by saying "using MapReduce directly is not allowed", the interviewer meant that we have to give a detailed explanation about how the work is distributed among the network of computers, instead of just saying "the MapReduce framework will distribute and combine the work for us".

I don't think any of the suggested solutions are right. It is possible to imagine a situation where some url is ranked number 11 on every box, and so has very high visits overall, while every url in the top 10 on each individual box is seen nowhere else, so has low visits overall.

That said, I don't have any better ideas. This problem is hard!

I understand that with the given constraints it is not possible to get a trivial solution. but i thik we have to consider the senario where one url in the actual 10 top urls is visited by most of the machines but only a few times that will keep it out of the individual top 10 lists of each machine.

does that sound right?

You could possibly continually request the top urls until you have ten where the smallest value in the top10 list is not more then the highest value in any server using paginated sets. For example,:

page = 1;
max=10;
while(top.length <max || top.smallest < highest && page <maxpages){
        highest = -1;
	for(server in farm){
            temp = getTop(server,max, page);
            merge(top, temp, max);
            if(highest < temp.top){
               highest = temp.top;
           }
        }
      page++;
}

Merge would keep no more than max values based on unique url and highest visits.

I would recursively group the servers into sets of two and aggregate the URL's .
Suppose there are 6 servers.
a) Group s1s2, s3s4, s5s6.
b) We know the entire map cannot be transmitted. Find out a safe message size for the network. Supposing n. Break the map in s1 into n size chunks and send over to s2.
Similarly from s3-s4, s5-s6.
c) This is the tricky part. The ques says we cannot do map-reduce directly. Does that mean map-reduce on the entire set? Is it allowed for individual machines? But it will be silly to solve this without ever getting a count of the URLs. So if map reduce is not allowed, write a procedure to sort the urls and track each one's count. This is done in s2, s4 and s6.
d) Now, again group the machines. This time {s2s4}, {s6}.
e) Repeat b, c . We will have s4, s6 left. Transmit from s4-s6 and perform a final count.

Lets assume, we can store count of k urls in the central machine. If the corpus is of size n, we get the top n/k elems from each server and send it to our central machine.

So, in our central machine, we have a set of k counters each initialized to 0 to start with. As we get our data from the stream, for every url, if it exists, we increment the counter. If not:
case1: if there is size in our central machine to add the url, we add it and set its count
case 2: if not, we delete the count of every url we have in the central machine. If there is any
url that has a count of 0, we delete it.

Now, the moment we hit case 2 above, we record the max_count_so_far and take a snapshot of the top 10 elements.

We process the next set of top n/k elements from the machines and for every max_count_so_far elements we take a snapshot of the top 10 elements.

At some point, say after we have 10 such snapshots, we find the final top 10 elements from the snapshots we have so far

On each server, we sort the urls based on their frequencies.
Say total log file lines across all servers are N. Number of servers is s. Capacity of server is k. Now split the ranked urls such that there are k/s elems per group on each server. Label each group , there will be total N/(k/s) ids.

Now, from this set of ids, we randomly select s ids (i.e s groups) such that each id doesnt occur more than threshold times on one machine. (To keep it simple lets say threshold=1).

Now, we employ the following algorithm on the central machine:
If there is size in our central machine to add the new group, we add it and set the count of each of the elems in the group. If not, we delete the count of every group/element in the group. We delete all urls that have a count of 0.

Now, one might argue that we might endup spending too much time on low frequency entries. We could then employ multiple iterations of elimination here. In the first pass, we can only consider frequencies that are above a certain threshold: say frequency above the median of the frequencies in each server. And divide those set of urls into groups. And consider random group from each server.

have a centralized counter. every time a new hit is recorded, recalc the centralized counter for that url with properly synced data structure. then compare it with the top10 hit stack by popping out the ones smaller than the current count result. keep updating the top 10 stack every time the new hit is recorded and you can always query the top 10 stack for the top 10 hits.

Lets say the number of servers is N. I think the solution requires 10 passes among the top ten scorers of each server. In each pass you can only identify one in the top ten. After each pass, the selected URL from the previous pass must be excluded from the evaluation in the subsequent passes, and the top ten scores from each server must be updated.

Think of the first pass. When all the top ten lists are considered, there must be at least one URL among these top tens which will land into the "real" top ten list. It is possible to generate scenarios where 9 of the real top ten list do not appear in the "current" top ten lists. But, there has to be at least one URL in the real top ten list which also appears in the 10N URLs that are collected from the servers' current top ten lists. Note that this is true only when N>=10. If N<10, all the real top ten list may not appear in the 10N URLs.

Also note that after collecting the 10N URLs, for some of them, you will have to ask the servers their frequencies so that you can sort the 10N URLs properly. Because, you want to sort the sum of visit frequencies of these URLs at each server.

After 10 such passes the real top ten list will emerge. The messaging complexity of each pass could be as bad as N^2 since you may have to collect the frequency values for each one of the 10N URLs. But, the computational complexity of each step is O(N) since you only need to find the max of 10N URLs.

If we were to design an algorithm that can find out not just 10 but the top K list. Then, we would be have to collect the top N scorers from each server .So, rather than 10N, we would bring together the top N^2 list at each pass. If K>N, this design is preferable.

the logic is similar to tetris game algorithm. Or generalize voting algorithm.

We partition the work by using a hash on the url. The hash gives an address to an aggregator who is responsible counting a subset (as defined by the hash) of urls. Then we crawl through the aggregators and fill the top-ten list. As each url is mapped to exactly one aggregator the crawler only has to maintain a map with ten entries.

Step 1) send table with aggregator addresses: say we use 10k aggregators
Step 2) on individual machine count all urls
Step 3) on individual machine iterate through counted urls, find hash-key, modulo 10k
gives you the address of responsible aggregator -> send to aggregator the count of said url.
Step 4) aggregators accumulate the counts
Step 5) initialize empty top-ten list, iterate through 10k aggregator machines
and maintain the top-ten list

We have a central server which will talk to each of the computers in the network containing the log files. Let's say there are n such computers.

Lets objectify each entry in the network of computers containing the log as <url, count, computer_id> where computer_id is the id of these individual computers.

class Entry {
	String URL,
	int count,
	String computer_id
}

We are going to solve this problem with map-reduce with minimal network data.

Map-step
-------------
Each of the log computers create a max heap (sorted on count) for all the URLs.

Reduce Step:
-----------------
The central server can ask a given computer, the top of the stack. Lets call this function as

function Entry getTopEntry(computer_id)

The central server can also ask a given computer to get the count for a given URL

function int getCount(computer_id, URL)

The central server has two heaps -
- one max-heap of size n (lets call this current_max_heap) containing the one (the top) entry from each log-computers. We also store the sum of all the entries' count in a variable and keep it updated.
- one min-heap of size k (lets call this k_min_heap) containing the current top k elements across all servers. At the end the reduce step, this heap will contain the required k entries

The idea:
Initialization: We initialize the current_max_heap with the first entry from each of the log computers (by polling their individual max_heaps).
We now get the top entry from the current_max_heap and add it to the k_min_heap. Before adding we need to update the entry's count so that it is equal to the count of that entry's URL across all the log-computers, which can be done by asking the count for a URL from all the log-computers) Every time an entry is removed from current_max_heap, we get the top entry from the computer's max-heap to which this entry belonged. We do it k times, so that the k_min_heap has k entries.
At any time, if the top element of the k_min_heap's count is less than the sum of the count's for element in current_max_heap), there is a possibility that the top URLs for all the computer's are same and when added-up they will be higher than this top-entry. So, we continue to move elements from current_max_heap to k_min_heap, till this condition is no longer true.

Here is the code running on the server:

// declare current_max_heap;
// declare k_min_heap;
int current_max_count = 0;
// Initialize the 
for (int i = 0; i < n; i++) {
	Entry entry = getTopEntry(computer_id);
	current_max_heap.add(entry);
	current_max_count += entry.count;
}
do {
	Entry top = current_max_heap.poll();
	Entry new_top  = getTop(top.computer_id);
	current_max_heap.add(new_top);
	current_max_count = current_max_count - top.count + new_top.count;
	current_max_heap.add();
	for (int i = 0; i < n; i++) {
		entry.total_count += getCount(i, top.URL);
	}
	if (k_min_heap.size() < k) {
		k_min_heap.add(entry);
	}
	else {
		if (k_min_heap.peek().count < entry.count) {
			k_min_heap.poll();
			k_min_heap.add(entry);
		}
	}
while(k_min_heap.size() != k || k_min_heap.peek().count <= current_max_count);

Hi. My simple answer is in Swift 3 developed in an xcode Playground. You will notice my urls are simply the domain. You can expand that part. I code not because the comments on Careercup do not allow links at all...

// Imports
import Foundation



// Our Function to input the log files of visited Urls
func determineTopVisitedUrls( vistedUrls : [[String]]) -> [(String, Int)] {
    
    // Set initial Variables
    var urlVisits: [(url: String, visits: Int)] = []
    var urlFound : Bool = false
    var foundIndex : Int = Int()
    
    // Loop thru all files
    for file in vistedUrls {
        
        // Loop thru all urls in Log Files
        for url in file {
            
            // Search thru existing results before appending to the list...
            if urlVisits.count > 0 {
                for visitIndex in 0..<urlVisits.count {
                
                    if (urlVisits[visitIndex].url == url) {
                        
                        // Increase List at that URL,
                        foundIndex = visitIndex
                        urlVisits[visitIndex].visits += 1
                        urlFound = true
                        
                        //print("Working with url : \(url) which was found at \(foundIndex)")
                        break
                        
                    }
                }
            }
            
            // If not found, append to the end...
            if urlFound == false {
                
                // Append to the list with a 1
                urlVisits.append(( url, 1))
                //print("Appending url: \(url) to the array urlVisits(count: \(urlVisits.count))")
                
                
            // Else, sort the list if necessary
            } else {
                
                // print("Found url, increasing visits by +1 to total of \(urlVisits[foundIndex].visits) and foudnIndex: \(foundIndex)")
                
                // loop thru results from found location up only....
                while foundIndex > 0 {
                    
                    // print("foundIndex: \(foundIndex)")
                    
                    if urlVisits[foundIndex].visits > urlVisits[foundIndex - 1].visits {
                        
                        // Switch them in the placement
                        let tempUrlVisit : (url: String, visits: Int) = urlVisits[foundIndex]
                        urlVisits[foundIndex] = urlVisits[foundIndex - 1]
                        urlVisits[foundIndex - 1] = tempUrlVisit
                        
                    } else {
                        
                        break
                    }
                    
                    foundIndex -= 1
                    
                }
                
            }
            
            // Reset Values before next search and append
            urlFound = false
            
        }
    }
    
    // remove everything after topTen
    if urlVisits.count > 10 {
        
        for afterIndex in 10...urlVisits.count {
            
            urlVisits.remove(at: afterIndex)
            
        }
    }
    
    // Return Top Urls no greater then 10
    return urlVisits
    
}



// Hi, you will need to make the urls longer.  The comments can't have links so I couldn't provide the full long url with query strings and such... Please expand the urls at the end and beginning of each logFile
..


// Create 2 Log Files
var logFile1 : [String] = [ "espn", "youtube", "youtube", "apple", "apple", "mindcraft" ]


var logFile2 : [String] = [ "espn", "apple", "youtube", "youtube", "mindcraft" ]


// Create an Array of the Log Files
var allFiles : [[String]] = [ logFile1, logFile2 ]




// Call our Function for testing
var topTenUrls : [(String, Int)] = determineTopVisitedUrls(vistedUrls: allFiles)





// Print the results of the top Urls
print("The top \(topTenUrls.count) Urls are \(topTenUrls) ")

1) Use distributed caching like memcached
2) Have a central computer which will act as primary server for processing final results.
Add multiple secondary computer as backups incase primary goes down.
3) The central computer cluster updates a threshold for count of most visited urls.
We can either put the threshold in memcached or distribute it using an observer pattern where all
registered computers get the updated threshold.
4) Each computer calculates up the 10 max urls by visited count which are more than the
threshold and update them in memcached. This can be either scheduled or triggered when they are notified
in the observer pattern.
5) The central computer will iterate over all the data cached in memcached and calculates the top ten.

1. Central server asks which computer has highest top ten and receives that. Highest is defined as bigger tenths element (minimum of top ten).

2. Sever ask each computer for corresponding URLs, it just received, and computers remove those URLs from their list.

3. Server updates its top then with received URLs.

4. Repeat this until no computer has bigger top ten than server top ten divided by number of computers.

The easy solution to solve this kind of problem is MapReduce. Since MapReduce is not allowed, the other alternative is to sort the string (url) ==> int (visits) in each machine independently such that it is the increasing order of visits. Then each server can send their top 10 visited URLs, Visit mapping to one of the server.

This single server would receive data from all the others for the top 10 visited sites and do a merge to decide on the top 10 sites. ( A n-way merge sort of the URL vs visits).

edit: this is does not guarantee 100% correctness

The question says we can't use MapReduce directly because the amount of data is too large. However, the overall top 10 sites are *expected* to be in the top 10 of at least one of the computers.

Hence, we can sort the logs of each computer and emit the top 10 of each computer in the mapping phase. Then, the reduce phase aggregates the number of visits for each site. The result will be the top 10 of the aggregate.

What if we manage a MaxHeap of top 10 sites at each node separately, and a heap of size 10 in central unit. The central heap can be updated according to :

if(heap[0]<RecentlyReceievedData)
{
   heap[0]=RecentlyReceivedData;
   MaxHeap(heap,0);
}

Then at every pre-specified interval, we can request updates from the nodes whose responses are the content of their respective MaxHeap.

We could use a BFS, visit every node and add sum of the integers to the master map that has all the URL's checking, if the URL is already present, then just add the integer orelse create a new key and put it into the map. Then traverse through the map to find the top ten values or just sort descending and return the first 10 values.

There are N number of machines,
Determine top 10 in every machine.
Each machine transmits its top 10 to every other machine i.e (n-1)10 urls
This Implies each machine also recieves (n-1)10 urls

Process these and determine the top 10 in every machine.

Each machine now sends its Top 10 to a single place,(n*10)
The top 10 among these (n*10) will be the most visited URL's

I do not know if there is any way to avoid not replicating the data on each machine. But will N*(N-1)*10 url's be too much traffic for the network to handle, cos that will be the total number of replications required.

Other possible solutions I could think of was using P-S pattern to publish count of URL's periodically.

1. tag the nodes as n1, n2, n3..... nk
2. First n1 sorts its list of URLs to find top10
3. n1 sends this list to n2. n2 adds this list to the data set, and gets a top 10
4. now n2 sends its top10 (calculated in step 3) to n3
5. keep doing it till we reach nk

nk will have the cumulative top 10