2.1. Sort by Key

Algorithm that sorts the elements of a set of files and merges the partial results respecting the order.

First of all - Create a dataset

This step can be avoided if the dataset already exists.

If not, this code snipped creates a set of files with dictionary on each one generated randomly. Uses pickle.

def datasetGenerator(directory, numFiles, numPairs):
    import random
    import pickle
    import os
    if os.path.exists(directory):
        print("Dataset directory already exists... Removing")
        import shutil
        shutil.rmtree(directory)
    os.makedirs(directory)
    for f in range(numFiles):
        fragment = {}
        while len(fragment) < numPairs:
            fragment[random.random()] = random.randint(0, 1000)
        filename = 'file_' + str(f) + '.data'
        with open(directory + '/' + filename, 'wb') as fd:
            pickle.dump(fragment, fd)
        print('File ' + filename + ' has been created.')

numFiles = 2
numPairs = 10
directoryName = 'mydataset'
datasetGenerator(directoryName, numFiles, numPairs)

Dataset directory already exists... Removing
File file_0.data has been created.
File file_1.data has been created.

# Show the files that have been created
%ls -l $directoryName

total 8
-rw-r--r-- 1 user users 134 oct 24 15:05 file_0.data
-rw-r--r-- 1 user users 135 oct 24 15:05 file_1.data

Algorithm definition

import pycompss.interactive as ipycompss

import os
if 'BINDER_SERVICE_HOST' in os.environ:
    ipycompss.start(graph=True,
                    project_xml='../xml/project.xml',
                    resources_xml='../xml/resources.xml')
else:
    ipycompss.start(graph=True, monitor=1000)

********************************************************
**************** PyCOMPSs Interactive ******************
********************************************************
*          .-~~-.--.           ______         ______   *
*         :         )         |____  \       |____  \  *
*   .~ ~ -.\       /.- ~~ .      __) |          __) |  *
*   >       `.   .'       <     |__  |         |__  |  *
*  (         .- -.         )   ____) |   _    ____) |  *
*   `- -.-~  `- -'  ~-.- -'   |______/  |_|  |______/  *
*     (        :        )           _ _ .-:            *
*      ~--.    :    .--~        .-~  .-~  }            *
*          ~-.-^-.-~ \_      .~  .-~   .~              *
*                   \ \ '     \ '_ _ -~                *
*                    \`.\`.    //                      *
*           . - ~ ~-.__\`.\`-.//                       *
*       .-~   . - ~  }~ ~ ~-.~-.                       *
*     .' .-~      .-~       :/~-.~-./:                 *
*    /_~_ _ . - ~                 ~-.~-._              *
*                                     ~-.<             *
********************************************************
* - Starting COMPSs runtime...                         *
* - Log path : /home/user/.COMPSs/Interactive_15/
* - PyCOMPSs Runtime started... Have fun!              *
********************************************************

from pycompss.api.task import task
from pycompss.api.parameter import FILE_IN

@task(returns=list, dataFile=FILE_IN)
def sortPartition(dataFile):
    '''
    Reads the dataFile and sorts its content which is assumed to be a dictionary {K: V}
    :param path: file that contains the data
    :return: a list of (K, V) pairs sorted.
    '''
    import pickle
    import operator
    with open(dataFile, 'rb') as f:
        data = pickle.load(f)
    # res = sorted(data, key=lambda (k, v): k, reverse=not ascending)
    partition_result = sorted(data.items(), key=operator.itemgetter(0), reverse=False)
    return partition_result

@task(returns=list, priority=True)
def reducetask(a, b):
    '''
    Merges two partial results (lists of (K, V) pairs) respecting the order
    :param a: Partial result a
    :param b: Partial result b
    :return: The merging result sorted
    '''
    partial_result = []
    i = 0
    j = 0
    while i < len(a) and j < len(b):
        if a[i] < b[j]:
            partial_result.append(a[i])
            i += 1
        else:
            partial_result.append(b[j])
            j += 1
    if i < len(a):
        partial_result + a[i:]
    elif j < len(b):
        partial_result + b[j:]
    return partial_result

def merge_reduce(function, data):
    import sys
    if sys.version_info[0] >= 3:
        import queue as Queue
    else:
        import Queue
    q = Queue.Queue()
    for i in data:
        q.put(i)
    while not q.empty():
        x = q.get()
        if not q.empty():
            y = q.get()
            q.put(function(x, y))
        else:
            return x

MAIN

Parameters (that can be configured in the following cell): * datasetPath: The path where the dataset is (default: the same as created previously).

import os
import time
from pycompss.api.api import compss_wait_on

datasetPath = directoryName  # Where the dataset is
files = []
for f in os.listdir(datasetPath):
    files.append(datasetPath + '/' + f)

startTime = time.time()

partialSorted = []
for f in files:
    partialSorted.append(sortPartition(f))
result = merge_reduce(reducetask, partialSorted)

result = compss_wait_on(result)

print("Elapsed Time(s)")
print(time.time() - startTime)
import pprint
pprint.pprint(result)

Found task: sortPartition
Found task: reducetask
Elapsed Time(s)
3.075985908508301
[(0.09553823909709169, 410),
 (0.1050539352105847, 361),
 (0.1101668165151074, 380),
 (0.16707163864406294, 34),
 (0.23169922686345557, 234),
 (0.2514576491846777, 420),
 (0.28212150305417993, 824),
 (0.3355535709072125, 44),
 (0.37840542685156164, 671),
 (0.4116698728713306, 812),
 (0.44489092464317737, 558),
 (0.5078389792729182, 421),
 (0.6359848462288427, 751),
 (0.6944889844090878, 277),
 (0.7638837227992719, 686),
 (0.8018921430882744, 499),
 (0.9045260886433034, 430),
 (0.918459522461975, 927),
 (0.9620166095263026, 865)]

ipycompss.stop()

********************************************************
***************** STOPPING PyCOMPSs ********************
********************************************************
Checking if any issue happened.
Warning: some of the variables used with PyCOMPSs may
         have not been brought to the master.
********************************************************