如何优化MapReduce作业以实现高效的按行数据读取？

MapReduce是一种编程模型，用于处理和生成大数据集，它由两个主要阶段组成：Map阶段和Reduce阶段，在Map阶段，输入数据被分割成多个独立的块，然后每个块被映射到一个键值对，在Reduce阶段，所有具有相同键的键值对被组合在一起，并应用一个规约函数以生成最终结果。

以下是一个简单的MapReduce作业示例，按行读取文本文件并统计每行的单词数量：

1、我们需要定义一个Mapper函数，它将接收一行文本作为输入，并将其拆分为单词，对于每个单词，我们将输出一个键值对，其中键是单词本身，值是1。

def mapper(line):
    words = line.split()
    for word in words:
        yield (word, 1)

2、我们需要定义一个Reducer函数，它将接收一个键值对列表，其中键是单词，值是1，Reducer的任务是将所有这些1相加，得到每个单词的总计数。

from collections import defaultdict
def reducer(key, values):
    total_count = sum(values)
    return (key, total_count)

3、现在我们可以将这些函数组合在一起，创建一个MapReduce作业，我们将使用一个简单的文本文件作为输入，并输出每个单词及其出现次数。

def mapreduce(input_file, output_file):
    # Map阶段
    with open(input_file, 'r') as f:
        map_results = []
        for line in f:
            map_results.extend(list(mapper(line)))
    # Shuffle阶段（在这里我们假设已经完成）
    shuffled_results = {}
    for key, value in map_results:
        if key not in shuffled_results:
            shuffled_results[key] = []
        shuffled_results[key].append(value)
    # Reduce阶段
    with open(output_file, 'w') as f:
        for key, values in shuffled_results.items():
            result = reducer(key, values)
            f.write(f"{result[0]}: {result[1]}n")
运行MapReduce作业
input_file = "input.txt"
output_file = "output.txt"
mapreduce(input_file, output_file)

这个简单的MapReduce作业将按行读取输入文件input.txt，并将结果写入输出文件output.txt，每行的结果将包含一个单词及其在该行中出现的次数。