Skip to content

Base map reduce

framework3.base.base_map_reduce

MapReduceStrategy

Bases: ABC

Abstract base class for implementing Map-Reduce strategies.

This class defines the interface for Map-Reduce operations, providing a structure for implementing various distributed computing strategies.

Key Features
  • Abstract methods for map and reduce operations
  • Support for custom map and reduce functions
  • Method to stop the Map-Reduce process
Usage

To create a new Map-Reduce strategy, inherit from this class and implement all abstract methods. For example:

class SimpleMapReduce(MapReduceStrategy):
    def __init__(self):
        self.intermediate = []
        self.result = None

    def map(self, data: list, map_function: Callable) -> None:
        self.intermediate = [map_function(item) for item in data]

    def reduce(self, reduce_function: Callable) -> Any:
        self.result = reduce_function(self.intermediate)
        return self.result

    def stop(self) -> None:
        self.intermediate = []
        self.result = None

# Usage
mr = SimpleMapReduce()
data = [1, 2, 3, 4, 5]
mr.map(data, lambda x: x * 2)
result = mr.reduce(sum)
print(result)  # Output: 30
mr.stop()

Methods:

Name Description
map

Any, map_function: Callable) -> Any: Abstract method to perform the map operation.
reduce

Callable) -> Any: Abstract method to perform the reduce operation.
stop

Abstract method to stop the Map-Reduce process.
Note

This is an abstract base class. Concrete implementations should override all abstract methods to provide specific Map-Reduce functionality.

Source code in framework3/base/base_map_reduce.py 
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class MapReduceStrategy(ABC):
    """
    Abstract base class for implementing Map-Reduce strategies.

    This class defines the interface for Map-Reduce operations, providing a structure
    for implementing various distributed computing strategies.

    Key Features:
        - Abstract methods for map and reduce operations
        - Support for custom map and reduce functions
        - Method to stop the Map-Reduce process

    Usage:
        To create a new Map-Reduce strategy, inherit from this class and implement
        all abstract methods. For example:

        ```python
        class SimpleMapReduce(MapReduceStrategy):
            def __init__(self):
                self.intermediate = []
                self.result = None

            def map(self, data: list, map_function: Callable) -> None:
                self.intermediate = [map_function(item) for item in data]

            def reduce(self, reduce_function: Callable) -> Any:
                self.result = reduce_function(self.intermediate)
                return self.result

            def stop(self) -> None:
                self.intermediate = []
                self.result = None

        # Usage
        mr = SimpleMapReduce()
        data = [1, 2, 3, 4, 5]
        mr.map(data, lambda x: x * 2)
        result = mr.reduce(sum)
        print(result)  # Output: 30
        mr.stop()
        ```

    Methods:
        map(data: Any, map_function: Callable) -> Any:
            Abstract method to perform the map operation.
        reduce(reduce_function: Callable) -> Any:
            Abstract method to perform the reduce operation.
        stop() -> None:
            Abstract method to stop the Map-Reduce process.

    Note:
        This is an abstract base class. Concrete implementations should override
        all abstract methods to provide specific Map-Reduce functionality.
    """

    @abstractmethod
    def map(self, data: Any, map_function: Callable) -> Any:
        """
        Perform the map operation on the input data.

        This method should be implemented to apply the map_function to each element
        of the input data, potentially in a distributed manner.

        Args:
            data (Any): The input data to be processed.
            map_function (Callable): The function to be applied to each data element.

        Returns:
            Any: The result of the map operation, which could be a collection of
                 intermediate key-value pairs or any other suitable format.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def map(self, data: list, map_function: Callable) -> list:
                return [map_function(item) for item in data]
            ```
        """
        pass

    @abstractmethod
    def reduce(self, reduce_function: Callable) -> Any:
        """
        Perform the reduce operation on the mapped data.

        This method should be implemented to apply the reduce_function to the
        intermediate results produced by the map operation.

        Args:
            reduce_function (Callable): The function to be used for reducing the mapped data.

        Returns:
            Any: The final result of the Map-Reduce operation.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def reduce(self, reduce_function: Callable) -> Any:
                return reduce_function(self.intermediate_results)
            ```
        """
        pass

    @abstractmethod
    def stop(self) -> None:
        """
        Stop the Map-Reduce process and perform any necessary cleanup.

        This method should be implemented to halt the Map-Reduce operation,
        release resources, and reset the state if needed.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def stop(self) -> None:
                self.intermediate_results = []
                self.final_result = None
                # Additional cleanup code...
            ```
        """
        ...
map(data, map_function) abstractmethod

Perform the map operation on the input data.

This method should be implemented to apply the map_function to each element of the input data, potentially in a distributed manner.

Parameters:

Name Type Description Default
data Any

The input data to be processed.

 required
map_function Callable

The function to be applied to each data element.

 required

Returns:

Name Type Description
Any Any

The result of the map operation, which could be a collection of intermediate key-value pairs or any other suitable format.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def map(self, data: list, map_function: Callable) -> list:
    return [map_function(item) for item in data]
Source code in framework3/base/base_map_reduce.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@abstractmethod
def map(self, data: Any, map_function: Callable) -> Any:
    """
    Perform the map operation on the input data.

    This method should be implemented to apply the map_function to each element
    of the input data, potentially in a distributed manner.

    Args:
        data (Any): The input data to be processed.
        map_function (Callable): The function to be applied to each data element.

    Returns:
        Any: The result of the map operation, which could be a collection of
             intermediate key-value pairs or any other suitable format.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def map(self, data: list, map_function: Callable) -> list:
            return [map_function(item) for item in data]
        ```
    """
    pass
reduce(reduce_function) abstractmethod

Perform the reduce operation on the mapped data.

This method should be implemented to apply the reduce_function to the intermediate results produced by the map operation.

Parameters:

Name Type Description Default
reduce_function Callable

The function to be used for reducing the mapped data.

 required

Returns:

Name Type Description
Any Any

The final result of the Map-Reduce operation.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def reduce(self, reduce_function: Callable) -> Any:
    return reduce_function(self.intermediate_results)
Source code in framework3/base/base_map_reduce.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@abstractmethod
def reduce(self, reduce_function: Callable) -> Any:
    """
    Perform the reduce operation on the mapped data.

    This method should be implemented to apply the reduce_function to the
    intermediate results produced by the map operation.

    Args:
        reduce_function (Callable): The function to be used for reducing the mapped data.

    Returns:
        Any: The final result of the Map-Reduce operation.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def reduce(self, reduce_function: Callable) -> Any:
            return reduce_function(self.intermediate_results)
        ```
    """
    pass
stop() abstractmethod

Stop the Map-Reduce process and perform any necessary cleanup.

This method should be implemented to halt the Map-Reduce operation, release resources, and reset the state if needed.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def stop(self) -> None:
    self.intermediate_results = []
    self.final_result = None
    # Additional cleanup code...
Source code in framework3/base/base_map_reduce.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@abstractmethod
def stop(self) -> None:
    """
    Stop the Map-Reduce process and perform any necessary cleanup.

    This method should be implemented to halt the Map-Reduce operation,
    release resources, and reset the state if needed.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def stop(self) -> None:
            self.intermediate_results = []
            self.final_result = None
            # Additional cleanup code...
        ```
    """
    ...