2
\$\begingroup\$

I have been writing some code (see component parts here and here) that:

  1. Takes a very large JSON (15GB gzipped, ~10million records)
  2. Extracts the relevant parts of the JSON into a list of lists
  3. Creates a list of all contiguous n-gram sub-lists found in the array
  4. Creates a counter to count the frequency of each n-gram
  5. Output the Counter showing the most common occurrences

When I run the complete function on the full dataset, I get out of memory errors.

Please help me optimise this code. Am I just looking for too many sub-list combinations?

I was thinking of possibly chunking up the JSON, processing in parallel and then combining the counters at the end, but I have no idea how to implement parallel processing in IPython 2.7.

import json
import gzip
import csv
import time
from itertools import combinations
from collections import Counter

def json_seq(infile,seq_limit=-1,lower=0,increment=-1):
    ## Script takes in a journey layer JSON and creates a array of traversals,
    ## ignoring entry and exit nodes
    ## sample output: [['a','b','c'],['c','e','d','a','l'],['f',s']]

    ## infile : full path of JSON in GZip format
    ## seq_limit (optional) : integer value to only extract the first X traversals


    seq =[]
    j=0
    tot_len=0
    with gzip.open(infile) as f:
        for line in itertools.islice(f, lower, None):
            if j == seq_limit - lower or j == increment: ## only read in a certain number of traversals
                break
            jsonline = json.loads(line)[2]    # data is stored in this level of the JSON
            for i in range(0,len(jsonline)):
                jsonevent = jsonline[i][1]    # need to loop through this section of the JSON to extract relevant information
                if ('cat' in jsonevent) or ('dog' in jsonevent):  #certain data elements can be ignored to reduce the size of the list
                    continue
                seq.append(str(jsonevent)[0:])  # need to remove the first character 'u' from the JSON formatted string
            j = j + 1
            yield seq
            seq =[]

def subseq(sequences,ngram=None):
    ## Script takes an array of traversals and counts the number of times any
    ## contigious ngram appears across all traversals. The output is a counter of all sub-lists from the list of lists

    ## sequences : Array of traversals (from json_seq function)
    ## ngram (optional) : Restrict the code to only look for subsequences of length X

    if ngram == None:
        return Counter(seq[i:j] for seq in map(tuple, sequences) for i, j in combinations(range(len(seq) + 1), 2) if j - i > 1 and j - i < 7)
    else:
        return Counter(seq[i:i+int(ngram)] for seq in map(tuple, sequences) for i in range(len(seq) - int(ngram)))            

def test_function(infile,outfile=None,top_list=None,seq_limit=None,ngram=None):
    ## function takes JSON file and lists out distribution of all contigious
    ## subsequences. Returns a list of subsequences and frequencies.

    ## infile = full path of input file from JSON (Gzipped)
    ## outfile (optional) = full path of output text file for table of all subsequences and frequencies. Pipe delimited
    ## top_list (optional) = restrict output to top X subsequences only
    ## seq_limit (optional) = look at the first X sequences only
    ## ngram (optional) = search for X-gram's only

    seq =[]

    if top_list == None:
        for x in subseq(json_seq(infile,seq_limit),ngram).most_common():
            seq.append(x)
    else:
        for x in subseq(json_seq(infile,seq_limit),ngram).most_common(int(top_list)):
            seq.append(x)
    if outfile != None:
        with open(outfile,'wb+') as outputcsv:
            writer = csv.writer(outputcsv,delimiter='|')
            for key, count in seq:
                writer.writerow([key, count])
    yield seq       
###################################################################################
###################################################################################

infile = 'C:\Users\XXXX\XXXX\data_json.gz'
outfile = 'C:\Users\XXXX\XXXX\subsequence_output.txt'

print 'Start'
print time.ctime()
starttime = time.time()

list(test_function(infile,top_list=10,seq_limit=100000))

print 'End'
print time.ctime()
\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

This sounds like an opportunity for a map reduce algorithm. If your json file is just one big object that may not work. However, if the data just a multiline file with each line being another json object you could split it up very easily. Facebook and Mixpanel both export data this way to take advantage of the map reduce approach.

\$\endgroup\$
4
  • \$\begingroup\$ Yes, the JSON is a multi line file. How would I split up the algo to fit map reduce? I understand the concept but not how to use it \$\endgroup\$
    – ADatoo
    Oct 23, 2015 at 8:12
  • \$\begingroup\$ You need to decide how many "workers" you want processing the json at once and how many lines of json each worker gets. So start out with just 2 workers and give each 1000 lines. Each worker will process their lines and return back the finished results to your main program. Every time they finish 1000 lines the main program will decide if their are any lines left to process and then delegate those lines to the workers. Check out python's multiprocessing library docs.python.org/2/library/multiprocessing.html \$\endgroup\$ Oct 23, 2015 at 14:08
  • \$\begingroup\$ Interesting. Perhaps this is not the right place, but where would I split the function for map reduce? Do I do one map reduce over the test_function? Or do I do it on json_seq first and then on subseq? \$\endgroup\$
    – ADatoo
    Oct 23, 2015 at 19:57
  • \$\begingroup\$ All logic to process a row should be in the worker. The main app should just delegate rows to the workers. \$\endgroup\$ Oct 25, 2015 at 14:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.