dds package

Subpackages

• dds.core package
  • Submodules
  • dds.core.partition_generators module
    • BasicDataLoader
      • BasicDataLoader.retrieve_data()
    • DDSException
    • IPartitionGenerator
      • IPartitionGenerator.retrieve_data()
    • IteratorLoader
      • IteratorLoader.retrieve_data()
    • PickleLoader
      • PickleLoader.retrieve_data()
    • WorkerFileLoader
      • WorkerFileLoader.retrieve_data()
    • read_in_chunks()
    • read_lines()
  • dds.core.tasks module
    • distribute_partition()
    • map_and_save_pickle()
    • map_and_save_text_file()
    • map_partition()
    • reduce_dicts()
    • reduce_multiple()
    • task_collect_samples()
    • task_dict_to_list()
  • dds.core.utils module
    • default_hash()
  • Module contents

Submodules

dds.api module

PyCOMPSs DDS - API.

This file contains the DDS interface.

class dds.api.DDS

Bases: object

Distributed Data Set object.

collect(keep_partitions=False, future_objects=False)

Return all elements from all partitions.

Elements can be grouped by partitions by setting keep_partitions value as True.

Usage sample:

>>> dds = DDS().load(range(10), 2)
>>> dds.collect(True)
[[0, 1, 2, 3, 4], [5, 6, 7, 8, 9]]
>>> DDS().load(range(10), 2).collect()
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Parameters:

• keep_partitions – Keep Partitions?

• future_objects – Future objects?

Returns:

All elements from all partitions.

collect_as_dict()

Get (key,value) as { key: value }.

Usage sample:

>>> DDS().load([("a", 1), ("b", 1)]).collect_as_dict()
{'a': 1, 'b': 1}

Returns:

Dict.

combine_by_key(creator_func, combiner_func, merger_function, total_parts=-1)

Combine elements of each key.

Parameters:

• creator_func – To apply to the first element of the key. Takes only one argument which is the value from (k, v) pair. (e.g: v = list(v)).

• combiner_func – To apply when a new element with the same ‘key’ is found. It is used to combine partitions locally. Takes 2 arguments; first one is the result of ‘creator_func’ where the second one is a ‘value’ of the same ‘key’ from the same partition. (e.g: v1.append(v2)).

• merger_function – To merge local results. Basically takes two arguments -both are results of ‘combiner_func’. (e.g: list_1.extend(list_2)).

• total_parts – Number of partitions after combinations.

Returns:

Combined by key DDS object.

count()

Count everything up.

Usage sample:

>>> DDS().load(range(3), 2).count()
3

Returns:

Total number of elements.

count_by_key(as_dict=False)

Count by key.

Usage sample:

>>> DDS().load([("a", 100), ("a", 200)]).count_by_key(True)
{'a': 2}

Parameters:

as_dict – See ‘as_dict’ argument of ‘combine_by_key’.

Returns:

A new DDS with data set of list of tuples (element, occurrence).

count_by_value(arity=2, as_dict=True, as_fo=False)

Amount of each element on this data set.

Usage sample:

>>> first = DDS().load([0, 1, 2], 2)
>>> second = DDS().load([2, 3, 4], 3)
>>> dict(sorted(
...     first.union(second).count_by_value(as_dict=True).items()
... ))
{0: 1, 1: 1, 2: 2, 3: 1, 4: 1}

Parameters:

• arity – Tree depth.

• as_dict – As dictionary.

• as_fo – As future object.

Returns:

List of tuples (element, number).

distinct()

Get the distinct elements of this data set.

Usage sample:

>>> test = list(range(10))
>>> test.extend(list(range(5)))
>>> len(test)
15
>>> DDS().load(test, 5).distinct().count()
10

Returns:

New child DDS object with distinct elements.

filter(func)

Filter elements of this data set by applying a given function.

Usage sample:

>>> DDS().load(range(10), 5).filter(lambda x: x % 2).count()
5

Parameters:

func – Filtering function.

Returns:

New child DDS object filtered.

flat_map(func, *args, **kwargs)

Apply a function to each element of the dataset.

NOTE: Extends the derived element(s) if possible.

Usage sample:

>>> dds = DDS().load([2, 3, 4])
>>> sorted(dds.flat_map(lambda x: range(1, x)).collect())
[1, 1, 1, 2, 2, 3]

Parameters:

• func – A function that should return a list, tuple or another kind of iterable.

• args – Arguments.

• kwargs – Keyword arguments.

Returns:

New child DDS object.

flatten_by_key(func)

Reverse of combine by key.Flat (k, v) as (k, v1), (k, v2) etc.

In detail: (key, values) as (key, value1), (key, value2) …

Usage sample:

>>> DDS().load([('a',[1, 2]), ('b',[1])]).flatten_by_key(
...     lambda x: x
... ).collect()
[('a', 1), ('a', 2), ('b', 1)]

Parameters:

func – A function to parse values.

Returns:

Flattened by key.

foreach(func)

Apply a function to each element of this data set.

CAUTION: Does not return anything.

Parameters:

func – A void function.

Returns:

None

group_by_key(num_of_parts=-1)

Group values of each key in a single list.

A special and most used case of ‘combine_by_key’.

Usage sample:

>>> x = DDS().load([("a", 1), ("b", 2), ("a", 2), ("b", 4)])
>>> sorted(x.group_by_key().collect())
[('a', [1, 2]), ('b', [2, 4])]

Parameters:

num_of_parts – Number of parts.

Returns:

Grouped by key DDS object.

join(other, num_of_partitions=-1)

Join DDS objects.

Usage sample:

>>> x = DDS().load([("a", 1), ("b", 3)])
>>> y = DDS().load([("a", 2), ("b", 4)])
>>> sorted(x.join(y).collect())
[('a', (1, 2)), ('b', (3, 4))]

Parameters:

• other – Another DDS object.

• num_of_partitions – Number of partitions.

Returns:

Joined DDS objects.

key_by(func)

Create a (key,value) pair for each element where ‘key’ is f(value).

Usage sample:

>>> dds = DDS().load(range(3), 2)
>>> dds.key_by(lambda x: str(x)).collect()
[('0', 0), ('1', 1), ('2', 2)]

Parameters:

func – A Key Creator function which takes the element as a parameter and returns the key.

Returns:

List of (key, value) pairs.

keys()

Get keys.

Usage sample:

>>> DDS().load([("a", 1), ("b", 1)]).keys().collect()
['a', 'b']

Returns:

List of keys.

load(iterator, num_of_parts=10, paac=False)

Load and distribute the iterator on partitions.

Parameters:

• iterator – Partitions iterator.

• num_of_parts – Number of parts.

• paac – Partition as a collection.

Returns:

Self.

load_file(file_path, chunk_size=1024, worker_read=False)

Read file in chunks and save it onto partitions.

Usage sample:

>>> with open("test.file", "w") as testFile:
...     _ = testFile.write("Hello world!")
>>> DDS().load_file("test.file", 6).collect()
['Hello ', 'world!']

Parameters:

• file_path – A path to a file to be loaded.

• chunk_size – Size of chunks in bytes.

• worker_read – If reading the file in the worker (skips first bytes).

Returns:

Self.

load_files_from_dir(dir_path, num_of_parts=-1)

Read multiple files from a given directory.

Each file and its content is saved in a tuple in (‘file_path’, ‘file_content’) format.

Parameters:

• dir_path – A directory that all files will be loaded from.

• num_of_parts – Can be set to -1 to create one partition per file.

Returns:

Self.

load_pickle_files(dir_path)

Load serialized partitions from pickle files.

Parameters:

dir_path – Path to serialized partitions.

Returns:

Self.

load_text_file(file_name, chunk_size=1024, in_bytes=True, strip=True)

Load a text file into partitions with ‘chunk_size’ lines on each.

Usage sample:

>>> with open("test.txt", "w") as testFile:
...     _ = testFile.write("First Line! \n")
...     _ = testFile.write("Second Line! \n")
>>> DDS().load_text_file("test.txt").collect()
['First Line! ', 'Second Line! ']

Parameters:

• file_name – A path to a file to be loaded.

• chunk_size – Size of chunks in bytes.

• in_bytes – If chunk size is in bytes or in number of lines.

• strip – If line separators should be stripped from lines.

Returns:

Self.

map(func, *args, **kwargs)

Apply the given function to each element of the dataset.

Usage sample:

>>> dds = DDS().load(range(10), 5).map(lambda x: x * 2)
>>> sorted(dds.collect())
[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

Parameters:

• func – Function to apply.

• args – Arguments.

• kwargs – Keyword arguments.

Returns:

New child DDS object.

map_partitions(func)

Apply a function to each partition of this data set.

Usage sample:

>>> DDS().load(range(10), 5).map_partitions(
...     lambda x: [sum(x)]
... ).collect(True)
[[1], [5], [9], [13], [17]]

Parameters:

func – Function to apply.

Returns:

New child DDS object.

map_values(func)

Apply a function to each value of (k, v) element of this data set.

Usage sample:

>>> DDS().load([("a", 1), ("b", 1)]).map_values(
...     lambda x: x+1
... ).collect()
[('a', 2), ('b', 2)]

Parameters:

func – A function which takes ‘value’s as parameter.

Returns:

New DDS.

num_of_partitions()

Get the total amount of partitions.

Usage sample:

>>> DDS().load(range(10), 5).num_of_partitions()
5

Returns:

Number of partitions.

partition_by(partitioner_func=<function default_hash>, num_of_partitions=-1)

Create partitions by a Partition Func.

Usage sample:

>>> dds = DDS().load(range(6)).map(lambda x: (x, x))
>>> dds.partition_by(num_of_partitions=3).collect(True)
[[(0, 0), (3, 3)], [(1, 1), (4, 4)], [(2, 2), (5, 5)]]

Parameters:

• partitioner_func – A Function distribute data on partitions based on for example, hash function.

• num_of_partitions – Number of partitions to be created.

Returns:

Partitions.

reduce(func, initial='COMPSS_DEFAULT_VALUE_TO_BE_USED_AS_A_MARKER', arity=-1)

Reduce the whole data set.

Usage sample:

>>> DDS().load(range(10), 5).reduce((lambda b, a: b + a) , 100)
145

Parameters:

• func – A reduce function which should take two parameters as inputs and return a single result which will be sent to itself again.

• initial – Initial value for reducer which will be used to reduce the first element with.

• arity – Tree depth.

Returns:

Reduced result (inside a DDS if necessary).

reduce_by_key(func)

Reduce values for each key.

Usage sample:

>>> DDS().load([("a",1), ("a",2)]).reduce_by_key(
...     (lambda a, b: a+b)
... ).collect()
[('a', 3)]

Parameters:

func – a reducer function which takes two parameters and returns one.

Returns:

Reduced values.

save_as_pickle(path)

Save partitions of this DDS as pickle files.

Each partition is saved as a separate file for the sake of parallelism.

:param path:Destination file path. :return: None.

save_as_text_file(path)

Save string representations of DDS elements as text files.

This saving creates one file per partition.

Parameters:

path – Destination file path.

Returns:

None.

sort_by_key(ascending=True, num_of_parts=None, key_func=<function DDS.<lambda>>)

Sort by key.

Parameters:

• ascending – Ascending.

• num_of_parts – Number of parts.

• key_func – Key function.

Returns:

Sorted by key DDS object.

sum()

Sum everything up.

Usage sample:

>>> DDS().load(range(3), 2).sum()
3

Returns:

The sum of everything.

take(num)

Take the first num elements of DDS.

Parameters:

num – Number of elements to be retrieved.

Returns:

First elements of DDS.

union(*args)

Combine this data set with some other DDS data.

Usage sample:

>>> first = DDS().load([0, 1, 2, 3, 4], 2)
>>> second = DDS().load([5, 6, 7, 8, 9], 3)
>>> first.union(second).count()
10

Parameters:

args – Arbitrary amount of DDS objects.

Returns:

New DDS object combining two DDS objects.

values()

Get values.

Usage sample:

>>> DDS().load([("a", 1), ("b", 2)]).values().collect()
[1, 2]

Returns:

List of values.

Module contents

Main Quantum Distributed Library package.