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 135 nov  7 11:09 file_0.data
-rw-r--r-- 1 user users 133 nov  7 11:09 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_17/
* - 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.8698995113372803
[(0.041892723435254076, 357),
 (0.06359546469536281, 14),
 (0.08331557020368874, 878),
 (0.10278403432902072, 880),
 (0.12426682077744233, 630),
 (0.1517902880938985, 337),
 (0.16148801796133683, 356),
 (0.3738492069799837, 194),
 (0.3936177040756046, 529),
 (0.40150631691932415, 258),
 (0.4314880044253141, 176),
 (0.4397589460812781, 894),
 (0.5282304183357669, 391),
 (0.5342789370009435, 971),
 (0.8090653537847396, 621),
 (0.8430013017668287, 166),
 (0.8726915372538349, 908),
 (0.9031628077963247, 744)]

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.
********************************************************