我有一个数据集,其密钥由 3 部分组成:a、b 和 c。在我的映射器中,我想发出键为“a”、值为“a、b、c”的记录
如何为从 Hadoop 中的映射器检测到的每个“a”发出 10% 的总记录?是否应该考虑将先前 Map-Reduce 作业中每个“a”看到的记录总数保存在临时文件中?
我有一个数据集,其密钥由 3 部分组成:a、b 和 c。在我的映射器中,我想发出键为“a”、值为“a、b、c”的记录
如何为从 Hadoop 中的映射器检测到的每个“a”发出 10% 的总记录?是否应该考虑将先前 Map-Reduce 作业中每个“a”看到的记录总数保存在临时文件中?
使用此 java 代码随机选择 10%:
import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
public class RandomSample {
public static class Map extends Mapper<LongWritable, Text, Text, Text> {
private Text word = new Text();
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
if (Math.random()<0.1)
context.write(value,null);
else
context.write(null,null);
context.write(value,null);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = new Job(conf, "randomsample");
job.setJarByClass(RandomSample.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setNumReduceTasks(0);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.waitForCompletion(true);
}
}
并使用这个 bash 脚本来运行它
echo "Running Job"
hadoop jar RandomSample.jar RandomSample $1 tmp
echo "copying result to local path (RandomSample)"
hadoop fs -getmerge tmp RandomSample
echo "Clean up"
hadoop fs -rmr tmp
例如,如果我们命名脚本random_sample.sh
,要从文件夹中选择 10% /example/
,只需运行
./random_sample.sh /example/
如果你想要接近 10%,你可以使用 Random。这是一个映射器的例子:
public class Test extends Mapper<LongWritable, Text, LongWritable, Text> {
private Random r = new Random();
@Override
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
if (r.nextInt(10) == 0) {
context.write(key, value);
}
}
}