我很好奇 mapreduce 作业是否在单台机器上使用多线程。比如我的hadoop集群有10台服务器,默认情况下,如果输入文件足够大,就会有10个mappers。单个映射器是否在单台机器上使用多线程?
最佳答案
Is the single mapper using multiple threading in a single machine?
是的。 Mapreduce 作业可以使用多线程映射器(多线程或线程池运行 map
方法)。
我已经为Map only Hbase jobs使用了更好的 CPU 利用率...
MultiThreadedMapper
非常适合如果您的操作是高度 CPU 密集型的,可以提高速度。
mapper 类应该扩展 org.apache.hadoop.mapreduce.lib.map.MultithreadedMapper
而不是常规的 org.apache.hadoop.mapreduce.Mapper
。
The
Multithreadedmapper
has a different implementation of run() method. like below.run(org.apache.hadoop.mapreduce.Mapper.Context context)
Run the application's maps using a thread pool.
您可以通过以下方式设置 MultiThreadedMapper
映射器中的线程数
MultithreadedMapper.setNumberOfThreads(n);
或者您可以设置从属性文件加载的属性 mapred.map.multithreadedrunner.threads = n
并使用上面的 setter 方法(基于每个作业)来控制 cpu 密集度较低的作业。
这样做的影响,您可以在 mapreduce 计数器中看到,特别是与 CPU 相关的计数器。
Example Code snippet of MultithreadedMapper implementation :
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.map.MultithreadedMapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import java.io.IOException;
import java.util.regex.Pattern;
public class MultithreadedWordCount {
// class should be thread safe
public static class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
public static enum PREPOST { SETUP, CLEANUP }
@Override()
protected void setup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws java.io.IOException, java.lang.InterruptedException {
// will be called several times
context.getCounter(PREPOST.SETUP).increment(1);
}
@Override
protected void map(LongWritable key, Text value,
Context context) throws IOException, InterruptedException {
String[] words = value.toString().toLowerCase().split("[\\p{Blank}[\\p{Punct}]]+");
for (String word : words) {
context.write(new Text(word), new LongWritable(1));
}
}
@Override()
protected void cleanup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws java.io.IOException, InterruptedException {
// will be called several times
context.getCounter(PREPOST.CLEANUP).increment(1);
}
}
public static class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
@Override
protected void reduce(Text key, Iterable<LongWritable> values, Context context
) throws IOException, InterruptedException {
long sum = 0;
for (LongWritable value: values) {
sum += value.get();
}
context.write(key, new LongWritable(sum));
}
}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Job job = new Job();
job.setJarByClass(WordCount.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
MultithreadedMapper.setMapperClass(job, MultithreadedWordCount.WordCountMapper.class);
MultithreadedMapper.setNumberOfThreads(job, 10);
job.setMapperClass(MultithreadedMapper.class);
job.setCombinerClass(MultithreadedWordCount.WordCountReducer.class);
job.setReducerClass(MultithreadedWordCount.WordCountReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
/* begin defaults */
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
/* end defaults */
job.waitForCompletion(true);
}
}
关于multithreading - Mapreduce作业是否使用多线程,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39898334/