Skip to content

taps.apps.mapreduce

taps/apps/mapreduce.py

MapreduceApp 

MapreduceApp(
    data_dir: Path,
    map_tasks: int | None = None,
    generate: bool = False,
    generated_files: int = 10,
    generated_words: int = 10000,
)

Mapreduce application.

Parameters:

  • data_dir (Path) –

    Text file directory. Either contains existing text files (including in subdirectories) or will be used to store the randomly generated files.

  • map_tasks (int | None, default: None ) –

    Number of map tasks. If None, one map task is generated per text file. Otherwise, files are evenly distributed across the map tasks.

  • generate (bool, default: False ) –

    Generate random text files for the application.

  • generated_files (int, default: 10 ) –

    Number of text files to generate.

  • generated_words (int, default: 10000 ) –

    Number of words per text file to generate.

Source code in taps/apps/mapreduce.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def __init__(
    self,
    data_dir: pathlib.Path,
    map_tasks: int | None = None,
    generate: bool = False,
    generated_files: int = 10,
    generated_words: int = 10_000,
) -> None:
    self.generate = generate
    self.data_dir = data_dir

    if self.generate:
        files = generate_files(data_dir, generated_files, generated_words)
        logger.log(APP_LOG_LEVEL, f'Generated {len(files)} in {data_dir}')
    else:
        files = [f for f in data_dir.glob('**/*') if f.is_file()]
        logger.log(APP_LOG_LEVEL, f'Found {len(files)} in {data_dir}')

    self.files = files
    self.map_tasks = len(self.files) if map_tasks is None else map_tasks

close() 

close() -> None

Close the application.

Source code in taps/apps/mapreduce.py 
138
139
140
141
142
143
144
145
def close(self) -> None:
    """Close the application."""
    if self.generate:
        shutil.rmtree(self.data_dir)
        logger.log(
            APP_LOG_LEVEL,
            f'Removed generated files in {self.data_dir}',
        )

run() 

run(engine: Engine, run_dir: Path) -> None

Run the application.

Parameters:

  • engine (Engine) –

    Application execution engine.

  • run_dir (Path) –

    Run directory.

Source code in taps/apps/mapreduce.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def run(self, engine: Engine, run_dir: pathlib.Path) -> None:
    """Run the application.

    Args:
        engine: Application execution engine.
        run_dir: Run directory.
    """
    map_futures = [
        engine.submit(map_task, *batch)
        for batch in _chunkify(self.files, self.map_tasks)
    ]
    logger.log(
        APP_LOG_LEVEL,
        f'Submitted {len(map_futures):,} map tasks over '
        f'{len(self.files):,} input files',
    )

    reduce_future = engine.submit(reduce_task, *map_futures)
    logger.log(APP_LOG_LEVEL, 'Submitted reduce task')

    word_counts = reduce_future.result()
    logger.log(APP_LOG_LEVEL, 'Reduce task finished')

    most_common_words = word_counts.most_common(10)
    logger.log(
        APP_LOG_LEVEL,
        f'{len(most_common_words)} most frequent words:',
    )
    for word, count in most_common_words:
        logger.log(APP_LOG_LEVEL, f'{word} ({count:,})')

    logger.log(
        APP_LOG_LEVEL,
        f'Total number of words: {sum(word_counts.values()):,}',
    )

map_task() 

map_task(*files: Path) -> Counter[str]

Count words in files.

Source code in taps/apps/mapreduce.py 
21
22
23
24
25
26
27
28
def map_task(*files: pathlib.Path) -> Counter[str]:
    """Count words in files."""
    counts: Counter[str] = Counter()
    for file in files:
        with open(file, errors='ignore') as f:
            for line in f:
                counts.update(line.split())
    return counts

reduce_task() 

reduce_task(*counts: Counter[str]) -> Counter[str]

Combine word counts.

Source code in taps/apps/mapreduce.py 
31
32
33
34
35
36
def reduce_task(*counts: Counter[str]) -> Counter[str]:
    """Combine word counts."""
    total: Counter[str] = Counter()
    for count in counts:
        total.update(count)
    return total

generate_word() 

generate_word(
    word_min_length: int, word_max_length: int
) -> str

Generate a random word.

Source code in taps/apps/mapreduce.py 
39
40
41
42
def generate_word(word_min_length: int, word_max_length: int) -> str:
    """Generate a random word."""
    length = random.randint(word_min_length, word_max_length)
    return ''.join(random.choices(string.ascii_lowercase, k=length))

generate_text() 

generate_text(
    word_count: int,
    word_min_length: int,
    word_max_length: int,
) -> str

Generate a paragraph with the specified number of words.

Source code in taps/apps/mapreduce.py 
45
46
47
48
49
50
51
52
53
54
def generate_text(
    word_count: int,
    word_min_length: int,
    word_max_length: int,
) -> str:
    """Generate a paragraph with the specified number of words."""
    return ' '.join(
        generate_word(word_min_length, word_max_length)
        for _ in range(word_count)
    )

generate_files() 

generate_files(
    directory: Path,
    file_count: int,
    words_per_file: int,
    *,
    min_word_length: int = 2,
    max_word_length: int = 10
) -> list[Path]

Generate text files with random text.

Parameters:

  • directory (Path) –

    Directory to write the files to.

  • file_count (int) –

    Number of files to generate.

  • words_per_file (int) –

    Number of words per file.

  • min_word_length (int, default: 2 ) –

    Minimum character length of randomly generated words.

  • max_word_length (int, default: 10 ) –

    Maximum character length of randomly generated words.

Returns:

  • list[Path] –

    List of generated files.

Raises:

Source code in taps/apps/mapreduce.py 
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
def generate_files(
    directory: pathlib.Path,
    file_count: int,
    words_per_file: int,
    *,
    min_word_length: int = 2,
    max_word_length: int = 10,
) -> list[pathlib.Path]:
    """Generate text files with random text.

    Args:
        directory: Directory to write the files to.
        file_count: Number of files to generate.
        words_per_file: Number of words per file.
        min_word_length: Minimum character length of randomly generated words.
        max_word_length: Maximum character length of randomly generated words.

    Returns:
        List of generated files.

    Raises:
        ValueError: if `directory` is not empty.
    """
    if directory.is_dir() and any(directory.iterdir()):
        raise ValueError('Directory {directory} is not empty')

    directory.mkdir(parents=True, exist_ok=True)

    files = []
    for i in range(file_count):
        filename = (directory / f'{i}.txt').resolve()
        text = generate_text(words_per_file, min_word_length, max_word_length)
        with open(filename, 'w') as f:
            f.write(text)
        files.append(filename)

    return files