Hadoop大数据处理框架教程

本文集文档发布于灏天文库

3.4 MapReduce 高级特性与优化

文档摘要

3.4 MapReduce 高级特性与优化 3.4 MapReduce 高级特性与优化 3.4.1 Combiner：本地聚合，减少数据传输 特性详解： Combiner，也称为“合并器”，是MapReduce框架中的一个可选组件，它位于Mapper和Reducer之间。其主要作用是在Mapper端输出数据的基础上，先进行一次本地聚合，再将聚合后的结果发送给Reducer。Combiner本质上是一个Reducer，但它运行在Mapper节点上，针对每个Mapper的输出结果进行局部聚合。 为什么要使用Combiner？ MapReduce作业的瓶颈往往在于网络带宽。Mapper的输出结果需要通过网络传输到Reducer，如果Mapper产生的数据量巨大，网络传输就会成为性能瓶颈。

3.4 MapReduce 高级特性与优化

3.4 MapReduce 高级特性与优化

3.4.1 Combiner：本地聚合，减少数据传输

特性详解：

Combiner，也称为“合并器”，是MapReduce框架中的一个可选组件，它位于Mapper和Reducer之间。其主要作用是在Mapper端输出数据的基础上，先进行一次本地聚合，再将聚合后的结果发送给Reducer。Combiner本质上是一个Reducer，但它运行在Mapper节点上，针对每个Mapper的输出结果进行局部聚合。

为什么要使用Combiner？

MapReduce作业的瓶颈往往在于网络带宽。Mapper的输出结果需要通过网络传输到Reducer，如果Mapper产生的数据量巨大，网络传输就会成为性能瓶颈。Combiner的出现正是为了解决这个问题。通过在Mapper端进行本地聚合，Combiner可以显著减少Mapper输出到Reducer的数据量，从而降低网络传输的开销，提升作业的整体性能。

Combiner的工作流程：

  1. Mapper输出数据： Mapper任务执行完毕后，会生成键值对 (<key, value>) 输出。

  2. Combiner本地聚合： 框架会将Mapper的输出数据按照Key进行分组，并对每个分组应用Combiner函数。Combiner函数与Reducer函数类似，接收相同Key的数据进行聚合处理，输出聚合后的结果 (<key, aggregated_value>)。

  3. 数据传输到Reducer： Combiner聚合后的数据才会被发送到Reducer节点。

  4. Reducer全局聚合： Reducer节点接收来自不同Mapper（经过Combiner处理）的数据，进行全局聚合，最终输出最终结果。

Combiner适用场景：

Combiner适用于那些聚合操作满足交换律和结合律的场景。例如：

  • 求和 (Sum)： (a + b) + c = a + (b + c)，(a + b) = (b + a)。Combiner可以先对Mapper输出的相同Key的值进行求和，再传输到Reducer进行全局求和。

  • 计数 (Count)： Combiner可以先在Mapper端统计每个Key出现的次数，再由Reducer进行全局计数。

  • 最大值/最小值 (Max/Min)： Combiner可以在Mapper端找到每个Key的最大值/最小值，Reducer再从这些局部最大值/最小值中找出全局最大值/最小值。

Combiner不适用场景：

  • 平均值 (Average)： 平均值不满足交换律和结合律。简单的本地平均值再进行全局平均值计算是不正确的。例如，需要先计算总和和总数，再在Reducer端计算平均值。

  • 中位数 (Median)： 中位数也不满足交换律和结合律。

代码实践：WordCount with Combiner

以经典的WordCount程序为例，我们来演示如何使用Combiner进行优化。

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class WordCountWithCombiner {
    public static class TokenizerMapper
            extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            String[] words = value.toString().toLowerCase().split("\\s+");
            for (String w : words) {
                word.set(w);
                context.write(word, one);
            }
        }
    }
    // Combiner 类，与 Reducer 类逻辑相同
    public static class IntSumCombiner
            extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }
    public static class IntSumReducer
            extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count with combiner");
        job.setJarByClass(WordCountWithCombiner.class);
        job.setMapperClass(TokenizerMapper.class);
        // 设置 Combiner 类
        job.setCombinerClass(IntSumCombiner.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
​

代码详解：

  • IntSumCombiner 类: 我们定义了一个 IntSumCombiner 类，其逻辑与 IntSumReducer 类完全相同。它继承自 Reducer，接收 Text (单词) 和 IntWritable (词频) 作为输入，并输出 Text 和 IntWritable。在 reduce 方法中，它对相同单词的词频进行求和。

  • job.setCombinerClass(IntSumCombiner.class);: 在 main 函数中，我们通过 job.setCombinerClass() 方法将 IntSumCombiner 类设置为作业的 Combiner。

Graph TD 图示 Combiner 工作流程：

图示解释：

  • Mapper Stage: 输入数据被分发到多个Mapper节点 (Mapper 1, Mapper 2)。每个Mapper输出键值对 (Mapper Output)。

  • Combiner Stage: 在每个Mapper节点上，Combiner (Combiner 1, Combiner 2) 对Mapper的输出进行本地聚合。

  • Shuffle & Reduce Stage: Combiner聚合后的数据经过 Partition 和 Shuffle 阶段，被分发到 Reducer 节点 (Reducer)。Reducer 进行全局聚合，最终输出结果 (Output Data)。

性能提升：

使用 Combiner 后，Mapper 输出到 Reducer 的数据量会大大减少，从而降低网络传输开销，缩短作业执行时间。尤其是在数据倾斜的情况下，Combiner 的作用更加明显。

3.4.2 Partitioner：控制数据分发，实现负载均衡

特性详解：

Partitioner，即“分区器”，负责将Mapper输出的中间结果划分到不同的Reducer。默认情况下，MapReduce使用 HashPartitioner，它根据Key的哈希值对Reducer的数量取模来决定数据分发到哪个Reducer。

为什么要使用Partitioner？

  • 负载均衡： 默认的 HashPartitioner 在Key分布均匀的情况下，可以实现较好的负载均衡，保证每个Reducer处理的数据量大致相同，避免出现“长尾效应”（某些Reducer任务执行时间过长）。

  • 数据分组： 在某些场景下，我们需要将具有相同特征的数据发送到同一个Reducer进行处理。例如，根据用户ID将用户行为数据分发到不同的Reducer进行用户行为分析。这时，就需要自定义Partitioner来实现特定的数据分组策略。

Partitioner的工作流程：

  1. Mapper输出数据： Mapper任务输出键值对 (<key, value>)。

  2. Partitioner分区： 框架调用 Partitioner 的 getPartition() 方法，根据 Mapper 输出的 Key 和 Reducer 的数量，计算出一个分区号 (partition ID)。

  3. 数据分发到Reducer： 框架根据分区号将数据发送到对应的 Reducer 节点。分区号相同的 Key-Value 对会被发送到同一个 Reducer。

自定义Partitioner：

当默认的 HashPartitioner 无法满足需求时，我们可以自定义 Partitioner 来实现更灵活的数据分区策略。自定义 Partitioner 需要继承 Partitioner<KEY, VALUE> 类，并重写 getPartition(KEY key, VALUE value, int numReduceTasks) 方法。

代码实践：自定义 Partitioner

假设我们有一个电商订单数据，包含订单ID、用户ID、商品ID和订单金额。我们希望根据用户ID的首字母将订单数据分发到不同的Reducer，以便对不同用户群体的订单进行分析。

import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.NullWritable;
public class UserIDPartitioner extends Partitioner<Text, NullWritable> {
    @Override
    public int getPartition(Text key, NullWritable value, int numReduceTasks) {
        if (numReduceTasks == 0) {
            return 0;
        }
        String userID = key.toString().split(",")[1]; // 假设用户ID在 Key 的第二个字段
        char firstLetter = userID.toUpperCase().charAt(0);
        if (firstLetter >= 'A' && firstLetter <= 'G') {
            return 0 % numReduceTasks; // 分区 0
        } else if (firstLetter >= 'H' && firstLetter <= 'N') {
            return 1 % numReduceTasks; // 分区 1
        } else if (firstLetter >= 'O' && firstLetter <= 'U') {
            return 2 % numReduceTasks; // 分区 2
        } else {
            return 3 % numReduceTasks; // 分区 3 (其他字母或特殊字符)
        }
    }
}
​

代码详解：

  • UserIDPartitioner 类: 我们定义了一个 UserIDPartitioner 类，继承自 Partitioner<Text, NullWritable>。这里假设 Mapper 的输出 Key 是 Text 类型，Value 是 NullWritable 类型（可以根据实际情况调整）。

  • getPartition() 方法: 重写 getPartition() 方法，根据用户ID的首字母进行分区。

    • 首先，从 Key 中提取用户ID（假设用户ID是逗号分隔的第二个字段）。

    • 获取用户ID的首字母，并转换为大写。

    • 根据首字母的范围，将数据划分到不同的分区 (0, 1, 2, 3)。

    • 使用 % numReduceTasks 确保分区号在有效的范围内。

在 Job 中设置自定义 Partitioner：

Job job = Job.getInstance(conf, "order analysis");
// ... (其他 Job 配置)
job.setPartitionerClass(UserIDPartitioner.class);
job.setNumReduceTasks(4); // 设置 Reducer 数量为 4
// ... (其他 Job 配置)
​

Graph TD 图示 Partitioner 工作流程：

图示解释：

  • Mapper Output: Mapper 任务输出的中间结果。

  • Partition Function (Partitioner): Partitioner 根据 Key 和 Reducer 数量，计算分区号。

  • Reducer 1, Reducer 2, ..., Reducer N+1: 数据根据分区号被分发到不同的 Reducer 节点进行处理。

负载均衡与数据倾斜：

自定义 Partitioner 可以帮助我们更好地控制数据分发，但同时也需要注意数据倾斜问题。如果分区策略不当，可能导致某些分区的数据量远大于其他分区，从而造成负载不均衡，影响作业性能。在设计 Partitioner 时，需要充分考虑数据的分布特性，尽量实现均匀的数据分发。

3.4.3 InputFormat 与 OutputFormat 的优化

InputFormat 优化：

  • 选择合适的 InputFormat: Hadoop 提供了多种 InputFormat，例如 TextInputFormat (文本文件), SequenceFileInputFormat (SequenceFile), AvroFileInputFormat (Avro文件) 等。选择与输入数据格式匹配的 InputFormat 可以提高数据读取效率。

  • Input Split 大小调整: InputFormat 将输入数据划分为多个 Input Split，每个 Input Split 会分配给一个 Mapper 任务处理。调整 Input Split 的大小可以影响 Mapper 的并行度和数据本地性。

    • 增大 Input Split 大小: 减少 Mapper 任务数量，降低任务调度开销，但可能降低并行度。适用于大文件处理，可以提高数据本地性。

    • 减小 Input Split 大小: 增加 Mapper 任务数量，提高并行度，但可能增加任务调度开销，降低数据本地性。适用于小文件处理，可以提高并行度。

    • 可以通过配置 mapreduce.input.fileinputformat.split.minsize 和 mapreduce.input.fileinputformat.split.maxsize 参数来调整 Input Split 大小。

  • Input Compression: 对输入数据进行压缩可以减少磁盘IO和网络传输开销。Hadoop 支持多种压缩格式，例如 Gzip, Bzip2, LZO, Snappy 等。选择合适的压缩格式可以平衡压缩比和解压速度。

OutputFormat 优化：

  • 选择合适的 OutputFormat: 类似于 InputFormat，选择与输出数据格式匹配的 OutputFormat 可以提高数据写入效率。常用的 OutputFormat 包括 TextOutputFormat (文本文件), SequenceFileOutputFormat (SequenceFile), AvroFileOutputFormat (Avro文件) 等。

  • Output Compression: 对输出数据进行压缩可以减少磁盘空间占用和后续数据处理的IO开销。可以通过 OutputFormat 配置输出数据的压缩格式。

  • 自定义 OutputFormat: 在某些场景下，可能需要将 MapReduce 的输出数据写入到特定的存储系统或数据库中。这时可以自定义 OutputFormat 来实现特定的数据输出逻辑。

3.4.4 Counter：监控与诊断

特性详解：

Counter，即“计数器”，是 MapReduce 框架提供的一种用于监控和诊断作业运行状态的机制。Counter 允许用户在 Mapper 和 Reducer 代码中定义和更新计数器，框架会在作业运行时收集和汇总这些计数器的值，最终在作业执行完成后展示出来。

Counter 的作用：

  • 监控作业进度： 通过定义计数器来统计处理的记录数、错误记录数等，可以实时监控作业的执行进度和数据质量。

  • 诊断作业问题： 通过计数器可以定位作业执行过程中的瓶颈和错误，例如数据倾斜、数据丢失等。

  • 性能分析： 通过计数器可以分析作业的性能瓶颈，例如Mapper 和 Reducer 的执行时间、IO 开销等。

Counter 的类型：

Hadoop Counter 主要分为两类：

  • 内置 Counter (Built-in Counters): 框架自动提供的计数器，用于监控 MapReduce 框架自身的运行状态，例如 Mapper 输入记录数、Reducer 输出记录数、文件系统IO统计、Map 和 Reduce 任务的CPU时间、内存使用量等。

  • 自定义 Counter (User-defined Counters): 用户在代码中定义的计数器，用于监控业务逻辑相关的指标。自定义 Counter 可以分为两类：

    • Map-Reduce Framework Counter: 由框架管理的计数器，例如 MAP_INPUT_RECORDS, REDUCE_OUTPUT_RECORDS 等。

    • User Counter: 用户自定义的计数器组和计数器名称。

代码实践：自定义 Counter

在 WordCount 程序中，我们统计处理的单词总数和空行数。

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Counter;
public class WordCountWithCounter {
    // 自定义 Counter 枚举
    public static enum WORD_COUNTER {
        TOTAL_WORDS,
        EMPTY_LINES
    }
    public static class TokenizerMapper
            extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            String line = value.toString();
            if (line.trim().isEmpty()) {
                // 统计空行
                Counter emptyLineCounter = context.getCounter(WORD_COUNTER.EMPTY_LINES);
                emptyLineCounter.increment(1);
                return;
            }
            String[] words = line.toLowerCase().split("\\s+");
            for (String w : words) {
                word.set(w);
                context.write(word, one);
                // 统计单词总数
                Counter totalWordCounter = context.getCounter(WORD_COUNTER.TOTAL_WORDS);
                totalWordCounter.increment(1);
            }
        }
    }
    // ... (Reducer 类与 WordCountWithCombiner 示例相同)
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count with counter");
        job.setJarByClass(WordCountWithCounter.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
​

代码详解：

  • WORD_COUNTER 枚举: 定义一个枚举类型 WORD_COUNTER，包含两个计数器：TOTAL_WORDS (单词总数) 和 EMPTY_LINES (空行数)。

  • context.getCounter() 方法: 在 Mapper 的 map 方法中，通过 context.getCounter(WORD_COUNTER.TOTAL_WORDS) 和 context.getCounter(WORD_COUNTER.EMPTY_LINES) 获取计数器对象。

  • counter.increment(1) 方法: 使用 counter.increment(1) 方法对计数器进行递增。

查看 Counter 值：

作业执行完成后，可以在 Hadoop Web UI 或命令行工具中查看 Counter 的值。在 Web UI 中，可以在 Job History 页面的 "Counters" 选项卡中查看。在命令行中，可以使用 hadoop job -history <job_id> 命令查看作业历史信息，其中包含 Counter 的值。

3.4.5 Distributed Cache：共享作业资源

特性详解：

Distributed Cache，即“分布式缓存”，允许用户将作业需要的文件（例如配置文件、字典文件、jar 包等）分发到集群中的所有节点，供 Mapper 和 Reducer 任务使用。这些文件会被缓存在节点的本地磁盘上，任务可以直接从本地磁盘读取，避免重复的网络传输，提高作业效率。

Distributed Cache 的作用：

  • 共享只读数据： 将作业需要的只读数据（例如字典文件、配置文件）缓存到本地，减少网络IO。

  • 代码共享： 将作业需要的 jar 包添加到 Distributed Cache，可以方便地在 Mapper 和 Reducer 中使用自定义的代码库。

  • 提高作业效率： 通过减少网络传输和磁盘IO，提高作业的执行效率。

使用 Distributed Cache 的方式：

  1. 上传文件到 HDFS： 首先将需要共享的文件上传到 HDFS。

  2. 配置 Distributed Cache： 在 Job 的配置中，使用 DistributedCache 类或 Configuration 对象添加需要缓存的文件。

  3. 任务中访问缓存文件： 在 Mapper 或 Reducer 的 setup() 方法中，通过 DistributedCache 类获取缓存文件的本地路径，并进行读取和使用。

代码实践：Distributed Cache 共享字典文件

假设我们有一个字典文件 dictionary.txt，包含一些敏感词汇。我们希望在 Mapper 中使用这个字典文件，过滤掉输入数据中的敏感词汇。

1. 上传字典文件到 HDFS：

hadoop fs -put dictionary.txt /user/hadoop/cache/dictionary.txt
​

2. 配置 Distributed Cache：

import java.io.IOException;
import java.net.URI;
import java.util.HashSet;
import java.util.Set;
import java.io.BufferedReader;
import java.io.FileReader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.filecache.DistributedCache;
public class SensitiveWordFilter {
    public static class FilterMapper
            extends Mapper<LongWritable, Text, Text, Text> {
        private Set<String> sensitiveWords = new HashSet<>();
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            try {
                // 获取 Distributed Cache 中的字典文件本地路径
                URI[] cacheFiles = DistributedCache.getCacheFiles(context.getConfiguration());
                if (cacheFiles != null && cacheFiles.length > 0) {
                    String filePath = cacheFiles[0].getPath(); // 假设只有一个缓存文件
                    BufferedReader reader = new BufferedReader(new FileReader(filePath));
                    String line;
                    while ((line = reader.readLine()) != null) {
                        sensitiveWords.add(line.trim());
                    }
                    reader.close();
                }
            } catch (IOException e) {
                System.err.println("Error reading cache file: " + e.getMessage());
            }
        }
        public void map(LongWritable key, Text value, Context context
        ) throws IOException, InterruptedException {
            String line = value.toString();
            String[] words = line.toLowerCase().split("\\s+");
            StringBuilder filteredLine = new StringBuilder();
            for (String word : words) {
                if (!sensitiveWords.contains(word)) {
                    filteredLine.append(word).append(" ");
                }
            }
            context.write(new Text("filtered_line"), new Text(filteredLine.toString().trim()));
        }
    }
    // ... (Reducer 类 - 可以直接使用 IdentityReducer)
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "sensitive word filter");
        job.setJarByClass(SensitiveWordFilter.class);
        job.setMapperClass(FilterMapper.class);
        job.setReducerClass(Reducer.class); // 使用 IdentityReducer
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 添加 Distributed Cache 文件
        DistributedCache.addCacheFile(new URI("/user/hadoop/cache/dictionary.txt#dictionary.txt"), conf);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
​

代码详解：

  • DistributedCache.addCacheFile() 方法: 在 main 函数中，使用 DistributedCache.addCacheFile() 方法将 HDFS 上的字典文件 /user/hadoop/cache/dictionary.txt 添加到 Distributed Cache。#dictionary.txt 部分指定了缓存文件在本地的软链接名称，方便在任务中访问。

  • DistributedCache.getCacheFiles() 方法: 在 Mapper 的 setup() 方法中，使用 DistributedCache.getCacheFiles() 方法获取 Distributed Cache 中的文件 URI 数组。

  • 读取缓存文件: 遍历 cacheFiles 数组，获取字典文件的本地路径，并使用 BufferedReader 读取文件内容，将敏感词汇加载到 sensitiveWords Set 中。

  • 过滤敏感词汇: 在 map() 方法中，对输入数据进行分词，并检查每个单词是否在 sensitiveWords Set 中。如果不在，则将其添加到 filteredLine 中。

Graph TD 图示 Distributed Cache 工作流程：

图示解释：

  • HDFS File: 需要共享的文件存储在 HDFS 上。

  • Distributed Cache: Distributed Cache 组件负责将文件分发到集群中的各个节点。

  • Node 1, Node 2, Node 3: 集群中的节点，运行 Mapper 或 Reducer 任务。

  • Local Cache Copy: 每个节点都会在本地磁盘上缓存 Distributed Cache 的文件副本。任务可以直接从本地磁盘读取缓存文件。

3.4.6 其他优化技巧

除了上述高级特性外，还有一些其他的 MapReduce 优化技巧可以提升作业性能：

  • 数据压缩 (Compression)： 在 MapReduce 作业的各个阶段（输入、中间结果、输出）使用数据压缩可以显著减少磁盘IO和网络传输开销。

    • 输入压缩: 对输入数据进行压缩，例如 Gzip, Bzip2, LZO, Snappy 等。

    • 中间结果压缩: 压缩 Mapper 的输出结果，可以使用 mapreduce.map.output.compress 和 mapreduce.map.output.compression.codec 参数配置。常用的中间结果压缩格式包括 Snappy, LZO 等，这些格式具有较快的压缩和解压速度。

    • 输出压缩: 对最终的输出数据进行压缩，可以使用 OutputFormat 的配置选项进行设置。

  • 推测执行 (Speculative Execution)： 启用推测执行后，框架会在某些任务执行速度明显慢于其他任务时，启动该任务的备份任务。一旦某个任务完成，框架会杀死其他相同的备份任务。推测执行可以有效地应对集群中节点性能不均或任务执行过程中出现短暂故障的情况，提高作业的整体执行速度。可以通过 mapreduce.map.speculative 和 mapreduce.reduce.speculative 参数开启推测执行。但需要注意，推测执行会消耗额外的资源，在资源紧张或任务本身就是性能瓶颈的情况下，可能反而会降低性能。

作者与出处
原作者： 灏天文库
来源：灏天文库
整理： 灏天文库整理
由灏天文库平台收录，内容或由平台用户上传，仅供学习交流
发布者: 作者: 灏天文库 转发
评论区 (0)
U
0/500