设为首页 收藏本站

运维网

 找回密码
 立即注册
查看: 665|回复: 0

[经验分享] hadoop学习2——DistributedCache的部分用法

[复制链接]
ienki

尚未签到
发表于 2016-12-11 10:13:25 | 显示全部楼层 |阅读模式
　　DistributedCache的部分用法。
　　调试代码：wordcount2.java

public class WordCount2 extends Configured implements Tool {
static Logger log = Logger.getLogger(WordCount2.class);
public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
static enum Counters {
INPUT_WORDS
}
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
private boolean caseSensitive = true;//是否区分大小写
private Set<String> patternsToSkip = new HashSet<String>();//替换使用的正则表达式
private long numRecords = 0;//数据量
private String inputFile;
public void configure(JobConf job) {
caseSensitive = job.getBoolean("wordcount.case.sensitive", true);
inputFile = job.get("map.input.file");
log.info("caseSensitive:" + job.get("wordcount.case.sensitive")
+ "inputFile:" + inputFile
+ "patterns:" + job.get("wordcount.skip.patterns"));
if (job.getBoolean("wordcount.skip.patterns", false)) {
log.info("传入参数wordcount.skip.patterns");
Path[] patternsFiles = new Path[0];
try {
//patternsFiles[0] = DistributedCache.getCacheFiles(job);//读取正则表达式路径（通过配置参数传递）
URI[] uris = DistributedCache.getCacheFiles(job);
patternsFiles = new Path[uris.length];
for(int i = 0; i < uris.length; i++){
Path path = new Path(uris.toString());
//Path path = new Path("D:/patterns.txt");
patternsFiles = path;
}
//log.info(uris[0].toString());
//patternsFiles = DistributedCache.getLocalCacheFiles(job);
//log.info(patternsFiles.length);
} catch (IOException ioe) {
System.err.println("Caught exception while getting cached files: "
+ StringUtils.stringifyException(ioe));
}
for (Path patternsFile : patternsFiles) {
parseSkipFile(patternsFile);
}
}
}
//提取文件中的正则表达式
private void parseSkipFile(Path patternsFile) {
log.info("提取文件中的正则表达式");
try {
//BufferedReader fis = new BufferedReader(new FileReader(patternsFile.toString()));
//BufferedReader fis = new BufferedReader(new FileReader("hdfs://192.168.100.228:9000/temp/p.dat"));
String pattern = null;
//while ((pattern = fis.readLine()) != null) {
//log.info("正则表达式:" + pattern);
//patternsToSkip.add(pattern);
//}
//读取hdfs中的模式表达式文件
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(patternsFile.toUri(), conf);
FSDataInputStream hdfsInStream = fs.open(patternsFile);
String s = "";
while (s != null) {
s = hdfsInStream.readLine();
if(s != null){
System.out.println(s);
patternsToSkip.add(s);
}
}
hdfsInStream.close();
//  fs.close();
log.info("正则表达式列表：" + patternsToSkip);
} catch (IOException ioe) {
System.err.println("Caught exception while parsing the cached file '"
+ patternsFile
+ "' : "
+ StringUtils.stringifyException(ioe));
}
}
public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
log.info("map 线程id：" + Thread.currentThread().getId());
String line = (caseSensitive) ? value.toString() : value.toString().toLowerCase();
for (String pattern : patternsToSkip) {
line = line.replaceAll(pattern, "");
}
StringTokenizer tokenizer = new StringTokenizer(line);
while (tokenizer.hasMoreTokens()) {
word.set(tokenizer.nextToken());
output.collect(word, one);
reporter.incrCounter(Counters.INPUT_WORDS, 1);
}
if ((++numRecords % 100) == 0) {
reporter.setStatus("Finished processing " + numRecords
+ " records " + "from the input file: " + inputFile);
}
}
}
public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
log.info("reduce 线程id：" + Thread.currentThread().getId());
int sum = 0;
while (values.hasNext()) {
sum += values.next().get();
}
output.collect(key, new IntWritable(sum));
}
}
public int run(String[] args) throws Exception {
JobConf conf = new JobConf(getConf(), WordCount2.class);
conf.setJobName("wordcount");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);
conf.setMapperClass(Map.class);
conf.setCombinerClass(Reduce.class);
conf.setReducerClass(Reduce.class);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
//设置正则表达式文件路径
DistributedCache.addCacheFile(new URI("/temp/p.dat"), conf);//向DistributedCache中add一个hdfs文件path
conf.setBoolean("wordcount.skip.patterns", true);

//List<String> other_args = new ArrayList<String>();
//for (int i = 0; i < args.length; ++i) {
//if ("-skip".equals(args)) {
//DistributedCache.addCacheFile(new Path(args[++i]).toUri(), conf);
//conf.setBoolean("wordcount.skip.patterns", true);
//} else {
//other_args.add(args);
//}
//}
FileInputFormat.setInputPaths(conf, new Path("/temp/in2"));
FileOutputFormat.setOutputPath(conf, new Path("/temp/out-" + String.valueOf(System.currentTimeMillis())));
JobClient.runJob(conf);
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new WordCount2(), args);
System.exit(res);
}
}
　　其中：
　　DistributedCache.addCacheFile(new URI("/temp/p.dat"), conf); //向DistributedCache中add一个hdfs文件path
　　patternsFiles[0] = DistributedCache.getCacheFiles(job); //读取正则表达式路径（通过配置参数传递）
表示了DistributedCache的基本用法。
　　使用DistributedCache可以在任务执行前传递一个URI路径，在map或reduce中可以使用DistributedCache.get*()拿到此路径对应的文件，这个文件可以是文档，jar包等。DistributedCache还提供直接把jar包加入classpath功能，利用这个功能，可以方便的使用第三方库。
　　输出日志：

12/02/09 17:28:50 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=JobTracker, sessionId=
12/02/09 17:28:51 INFO mapred.FileInputFormat: Total input paths to process : 2
12/02/09 17:28:51 INFO mapred.JobClient: Running job: job_local_0001
12/02/09 17:28:51 INFO mapred.FileInputFormat: Total input paths to process : 2
12/02/09 17:28:51 INFO mapred.MapTask: numReduceTasks: 1
12/02/09 17:28:51 INFO mapred.MapTask: io.sort.mb = 100
12/02/09 17:28:51 INFO mapred.MapTask: data buffer = 79691776/99614720
12/02/09 17:28:51 INFO mapred.MapTask: record buffer = 262144/327680
12/02/09 17:28:51 INFO test.WordCount2: caseSensitive:nullinputFile:hdfs://localhost:9000/temp/in2/t1.txtpatterns:true
12/02/09 17:28:51 INFO test.WordCount2: 传入参数wordcount.skip.patterns
12/02/09 17:28:51 INFO test.WordCount2: 提取文件中的正则表达式
12/02/09 17:28:51 INFO test.WordCount2: 正则表达式列表：[\! , \, , \. , to ]
\.
\,
\!
to
12/02/09 17:28:51 INFO test.WordCount2: map 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: map 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: map 线程id：22
12/02/09 17:28:51 INFO mapred.MapTask: Starting flush of map output
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:51 INFO mapred.MapTask: Finished spill 0
12/02/09 17:28:51 INFO mapred.TaskRunner: Task:attempt_local_0001_m_000000_0 is done. And is in the process of commiting
12/02/09 17:28:51 INFO mapred.LocalJobRunner: hdfs://localhost:9000/temp/in2/t1.txt:0+52
12/02/09 17:28:51 INFO mapred.TaskRunner: Task 'attempt_local_0001_m_000000_0' done.
12/02/09 17:28:51 INFO mapred.MapTask: numReduceTasks: 1
12/02/09 17:28:51 INFO mapred.MapTask: io.sort.mb = 100
12/02/09 17:28:51 INFO mapred.MapTask: data buffer = 79691776/99614720
12/02/09 17:28:51 INFO mapred.MapTask: record buffer = 262144/327680
12/02/09 17:28:51 INFO test.WordCount2: caseSensitive:nullinputFile:hdfs://localhost:9000/temp/in2/t2.txtpatterns:true
12/02/09 17:28:51 INFO test.WordCount2: 传入参数wordcount.skip.patterns
12/02/09 17:28:51 INFO test.WordCount2: 提取文件中的正则表达式
\.
\,
\!
to
12/02/09 17:28:51 INFO test.WordCount2: 正则表达式列表：[\! , \, , \. , to ]
12/02/09 17:28:51 INFO test.WordCount2: map 线程id：22
12/02/09 17:28:51 INFO test.WordCount2: map 线程id：22
12/02/09 17:28:51 INFO mapred.MapTask: Starting flush of map output
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO mapred.MapTask: Finished spill 0
12/02/09 17:28:52 INFO mapred.TaskRunner: Task:attempt_local_0001_m_000001_0 is done. And is in the process of commiting
12/02/09 17:28:52 INFO mapred.LocalJobRunner: hdfs://localhost:9000/temp/in2/t2.txt:0+35
12/02/09 17:28:52 INFO mapred.TaskRunner: Task 'attempt_local_0001_m_000001_0' done.
12/02/09 17:28:52 INFO mapred.LocalJobRunner:
12/02/09 17:28:52 INFO mapred.Merger: Merging 2 sorted segments
12/02/09 17:28:52 INFO mapred.Merger: Down to the last merge-pass, with 2 segments left of total size: 184 bytes
12/02/09 17:28:52 INFO mapred.LocalJobRunner:
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO test.WordCount2: reduce 线程id：22
12/02/09 17:28:52 INFO mapred.TaskRunner: Task:attempt_local_0001_r_000000_0 is done. And is in the process of commiting
12/02/09 17:28:52 INFO mapred.LocalJobRunner:
12/02/09 17:28:52 INFO mapred.TaskRunner: Task attempt_local_0001_r_000000_0 is allowed to commit now
12/02/09 17:28:52 INFO mapred.FileOutputCommitter: Saved output of task 'attempt_local_0001_r_000000_0' to hdfs://localhost:9000/temp/out-1328779730906
12/02/09 17:28:52 INFO mapred.LocalJobRunner: reduce > reduce
12/02/09 17:28:52 INFO mapred.TaskRunner: Task 'attempt_local_0001_r_000000_0' done.
12/02/09 17:28:52 INFO mapred.JobClient:  map 100% reduce 100%
12/02/09 17:28:52 INFO mapred.JobClient: Job complete: job_local_0001
12/02/09 17:28:52 INFO mapred.JobClient: Counters: 16
12/02/09 17:28:52 INFO mapred.JobClient:   FileSystemCounters
12/02/09 17:28:52 INFO mapred.JobClient:     FILE_BYTES_READ=67623
12/02/09 17:28:52 INFO mapred.JobClient:     HDFS_BYTES_READ=63479
12/02/09 17:28:52 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=64858
12/02/09 17:28:52 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=131732
12/02/09 17:28:52 INFO mapred.JobClient:   com.hadoop.test.WordCount2$Map$Counters
12/02/09 17:28:52 INFO mapred.JobClient:     INPUT_WORDS=16
12/02/09 17:28:52 INFO mapred.JobClient:   Map-Reduce Framework
12/02/09 17:28:52 INFO mapred.JobClient:     Reduce input groups=16
12/02/09 17:28:52 INFO mapred.JobClient:     Combine output records=16
12/02/09 17:28:52 INFO mapred.JobClient:     Map input records=5
12/02/09 17:28:52 INFO mapred.JobClient:     Reduce shuffle bytes=0
12/02/09 17:28:52 INFO mapred.JobClient:     Reduce output records=16
12/02/09 17:28:52 INFO mapred.JobClient:     Spilled Records=32
12/02/09 17:28:52 INFO mapred.JobClient:     Map output bytes=148
12/02/09 17:28:52 INFO mapred.JobClient:     Map input bytes=87
12/02/09 17:28:52 INFO mapred.JobClient:     Combine input records=16
12/02/09 17:28:52 INFO mapred.JobClient:     Map output records=16
12/02/09 17:28:52 INFO mapred.JobClient:     Reduce input records=16

　　测试数据是两个文件，执行了两个task，可以看出每次执行task时都会加载一次配置（读了两次配置文件）。

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表，该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有，请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意，请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息，一经发现立即删除。若您因此触犯法律，一切后果自负，我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集，我们仅提供一个展示、介绍、观摩学习的平台，我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责，亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏，不得用于商业或者其他用途，否则，一切后果均由您自己承担，我们对此不承担任何法律责任
7、如涉及侵犯版权等问题，请您及时通知我们，我们将立即采取措施予以解决
8、联系人Email：admin@iyunv.com 网址：www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址：https://www.yunweiku.com/thread-312670-1-1.html 上篇帖子： Hadoop-0.1版本FSNamesystem类笔记 下篇帖子： Hadoop学习第0天——配置分布式(译)
回复

使用道具 举报

返回列表
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群，最新一手资源尽在官方微信交流群！快快加入我们吧...

扫描微信二维码查看详情

客服 E-mail:iyunvcom@gmail.com

本站由青云提供云计算服务

运维网--中国最专业的运维工程师交流社区

豫ICP备20007574号-1 Copyright © 2012-2025

关于我们

发展历程

联系我们

关注我们

广告服务

广告案例

隐私条款

免责声明

用户登录

用户注册

版主守则

申请版主

手机版

金币获取

帮助中心

每日签到

Good Good

Study

Day Day

UP


客服E-mail:kefu@iyunv.com 客服QQ：1061981298


QQ群⑦：运维网交流群⑦ QQ群⑧：运维网交流群⑧ k8s群：运维网kubernetes交流群


提醒：禁止发布任何违反国家法律、法规的言论与图片等内容；本站内容均来自个人观点与网络等信息，非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来，其版权均归原作者及其网站所有，我们尊重他人的合法权益，如有内容侵犯您的合法权益，请及时与我们联系进行核实删除！



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表