前言：Hadoop当时我们弄时几乎没有什么中文文档。现在介绍的资料已经很多了，我就不再赘述。
业务描述：设定inputpath和ouputpath，根据访问日志分析某一个应用访问某一个API的总次数和总流量，统计后分别输出到两个文件中。注意：本案例我是改自阿里巴巴文初的那篇文章，他用的是0.17的版本，现在新的版本有很大的变动，我使用最新的0.20.2来改写了。
public class LogAnalysiser {
public static class MapClass
extends
org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, LongWritable>
{
public void map(LongWritable key, Text value,
OutputCollector<Text, LongWritable> output, Reporter reporter)
throws IOException
{
String line = value.toString();// 没有配置RecordReader，所以默认采用line的实现，key就是行号，value就是行内容
if (line == null || line.equals(""))
return;
String[] words = line.split(",");
if (words == null || words.length < 8)
return;
String appid = words[1];
String apiName = words[2];
LongWritable recbytes = new LongWritable(Long.parseLong(words[7]));
Text record = new Text();
record.set(new StringBuffer("flow::").append(appid)
.append("::").append(apiName).toString());
reporter.progress();
output.collect(record, recbytes);// 输出流量的统计结果，通过flow::作为前缀来标示。
record.clear();
record.set(new StringBuffer("count::").append(appid).append("::")
.append(apiName).toString());
output.collect(record, new LongWritable(1));// 输出次数的统计结果，通过count::作为前缀来标示
}
}
public static class PartitionerClass extends
org.apache.hadoop.mapreduce.Partitioner<Text, LongWritable>
{
public int getPartition(Text key, LongWritable value, int numPartitions)
{
if (numPartitions >= 2)// Reduce 个数，判断流量还是次数的统计分配到不同的Reduce
if (key.toString().startsWith("flow::"))
return 0;
else
return 1;
else
return 0;
}
/*public void configure(JobConf job) {
}*/
public void configure(Job job) {
}
}
public static class ReduceClass
extends
org.apache.hadoop.mapreduce.Reducer<Text, LongWritable, Text, LongWritable>
{
public void reduce(Text key, Iterator<LongWritable> values,
OutputCollector<Text, LongWritable> output, Reporter reporter)
throws IOException
{
Text newkey = new Text();
newkey.set(key.toString().substring(
key.toString().indexOf("::") + 2));
LongWritable result = new LongWritable();
long tmp = 0;
int counter = 0;
while (values.hasNext())// 累加同一个key的统计结果
{
tmp = tmp + values.next().get();
counter = counter + 1;// 担心处理太久，JobTracker长时间没有收到报告会认为TaskTracker已经失效，因此定时报告一下
if (counter == 1000)
{
counter = 0;
reporter.progress();
}
}
result.set(tmp);
output.collect(newkey, result);// 输出最后的汇总结果
}
}
public static class CombinerClass
extends
org.apache.hadoop.mapreduce.Reducer<Text, LongWritable, Text, LongWritable>
{
public void reduce(Text key, Iterator<LongWritable> values,
OutputCollector<Text, LongWritable> output, Reporter reporter)
throws IOException
{
LongWritable result = new LongWritable();
long tmp = 0;
while (values.hasNext())// 累加同一个key的统计结果
{
tmp = tmp + values.next().get();
}
result.set(tmp);
}
}
public static void main(String[] args)
{
try
{
run(args);
} catch (Exception e)
{
e.printStackTrace();
}
}
public static void run(String[] args) throws Exception
{
if (args == null || args.length < 2)
{
System.out.println("need inputpath and outputpath");
return;
}
String inputpath = args[0];
String outputpath = args[1];
String shortin = args[0];
String shortout = args[1];
if (shortin.indexOf(File.separator) >= 0)
shortin = shortin.substring(shortin.lastIndexOf(File.separator));
if (shortout.indexOf(File.separator) >= 0)
shortout = shortout.substring(shortout.lastIndexOf(File.separator));
SimpleDateFormat formater = new SimpleDateFormat("yyyy.MM.dd");
shortout = new StringBuffer(shortout).append("-")
.append(formater.format(new Date())).toString();
if (!shortin.startsWith("/"))
shortin = "/" + shortin;
if (!shortout.startsWith("/"))
shortout = "/" + shortout;
shortin = "/user/root" + shortin;
shortout = "/user/root" + shortout;
File inputdir = new File(inputpath);
File outputdir = new File(outputpath);
if (!inputdir.exists() || !inputdir.isDirectory())
{
System.out.println("inputpath not exist or isn't dir!");
return;
}
if (!outputdir.exists())
{
new File(outputpath).mkdirs();
}
Configuration conf = new Configuration();
Job job = new Job(conf, "analysis job");
job.setJarByClass(LogAnalysiser.class);
// JobConf conf = new JobConf(new Configuration(),
// LogAnalysiser.class);// 构建Config
FileSystem fileSys = FileSystem.get(conf);
fileSys.copyFromLocalFile(new Path(inputpath), new Path(shortin));// 将本地文件系统的文件拷贝到HDFS中
job.setJobName("analysisjob");
job.setOutputKeyClass(Text.class);// 输出的key类型，在OutputFormat会检查
job.setOutputValueClass(LongWritable.class); // 输出的value类型，在OutputFormat会检查
job.setMapperClass(MapClass.class);
job.setCombinerClass(CombinerClass.class);
job.setReducerClass(ReduceClass.class);
job.setPartitionerClass(PartitionerClass.class);
// job.set("mapred.reduce.tasks", "2");//老版本中的写法
// 强制需要有两个Reduce来分别处理流量和次数的统计,现在的版本中已经没有这个方法了
job.setNumReduceTasks(2);//新版本0.22.x中的方法
FileInputFormat.setInputPaths(job, shortin);// hdfs中的输入路径
FileOutputFormat.setOutputPath(job, new Path(shortout));// hdfs中输出路径
Date startTime = new Date();
System.out.println("Job started: " + startTime);
// JobClient.runJob(job);
Date end_time = new Date();
System.out.println("Job ended: " + end_time);
System.out.println("The job took "
+ (end_time.getTime() - startTime.getTime()) / 1000
+ " seconds.");
// 删除输入和输出的临时文件
fileSys.copyToLocalFile(new Path(shortout), new Path(outputpath));
fileSys.delete(new Path(shortin), true);
fileSys.delete(new Path(shortout), true);
}
}
//执行类
public class ExampleDriver {
public static void main(String argv[]){
ProgramDriver pgd = new ProgramDriver();
try {
pgd.addClass("analysislog", LogAnalysiser.class, "A map/reduce program that analysis log .");
pgd.driver(argv);
}
catch(Throwable e){
e.printStackTrace();
}
}
}
分享到：
相关推荐
这个"hadopp源代码存档"包含了Hadoop项目的完整源代码，供开发者深入理解其内部机制，进行二次开发或优化。 一、Hadoop核心组件 Hadoop主要由两个核心组件构成：Hadoop Distributed File System (HDFS) 和 ...
### 深入云计算 Hadoop源代码分析 #### 一、引言 随着大数据时代的到来，数据处理成为了各个领域中的关键技术之一。Hadoop作为一个开源的大数据处理框架，因其优秀的分布式计算能力，在业界得到了广泛的应用。...
《Hadoop集群程序设计与开发教材最终代码》这个压缩包文件是针对学习和理解Hadoop分布式计算框架的重要教学资源。Hadoop是Apache软件基金会开发的一个开源项目，它为大规模数据处理提供了一种分布式、容错性强的解决...
MapReduce是Hadoop处理大数据的主要计算模型，它将大规模数据处理任务分解为小的“映射”和“化简”任务，分别在集群的不同节点上并行执行。书中可能包含了各种MapReduce示例，如WordCount程序，用于统计文本中单词...
这是关于hadoop里面程序代码,有wordcount ,partition,onejoin, score,health,dedup,程序. 有.java,也有jar. 提示必须先装上hadoop才能运行
6. **Hadoop Shell命令**：Hadoop提供了丰富的命令行工具，如`hadoop fs`用于执行文件操作，`hadoop jar`用于运行MapReduce程序，`hadoop dfsadmin`用于管理系统设置。 7. **Hadoop生态系统**：除了核心组件，...
本项目将详细介绍如何在Hadoop环境中成功运行WordCount程序，以及涉及到的相关知识点。 首先，`Hadoop`是一个基于Java的框架，设计用来处理和存储大规模数据。它采用了分布式计算模型，即MapReduce，将大型任务分解...
在大数据处理领域，Hadoop API 是一个至关重要的工具集，它允许开发者编写程序来处理海量数据。这个压缩包文件 "shizhan_03_hadoop" 很可能包含了一些示例代码，展示了如何使用Hadoop API进行数据操作。现在，我们将...
很抱歉，但根据您提供的信息，标题和描述中提到的是"hadoop源代码归档"，而压缩包内的文件名称列表却包含了一系列法律学习资料，如刑法和民诉的思维导图、精讲卷以及口诀等，并没有提及任何与Hadoop源代码相关的内容...
阅读Hadoop源代码有助于理解其内部工作原理，从而优化自己的分布式应用程序。你可以学习错误处理机制、并发控制、网络通信等编程技巧。同时，通过设置断点和使用调试工具，你可以深入到代码执行的每个步骤，提升...
此外，还需要一个主程序来配置和提交作业。 **标签相关性**： - **Hadoop**：整个案例的基础框架，提供了分布式计算的能力。 - **大数据**：单词统计案例通常应用于处理海量文本数据，这是大数据场景的典型应用。 -...
很抱歉，但根据您提供的信息，标题和描述中提到的是"hadoop源代码版本归档"，而压缩包内的文件名称列表却包含了一系列法律考试复习资料，并没有与Hadoop相关的源代码或文档。这表明可能存在一个误解，因为压缩包里的...
- 应用管理器（ApplicationMaster）：每个应用程序有自己的ApplicationMaster，负责任务调度和监控。 - 节点管理器（NodeManager）：每个节点上运行的服务，负责容器的生命周期管理和资源隔离。 4. 分布式协调...
很抱歉，根据您提供的信息，压缩包中的文件名称列表似乎与Hadoop源代码归档的主题不相符，它们看起来像是视频文件而不是源代码或相关的技术文档。不过，关于"Hadoop源代码"这一主题，我可以提供一些核心知识点的详细...
通过这些代码，你可以学习到如何在Hadoop环境中编写、调试和运行应用程序，理解Hadoop的工作原理，并且可以动手实践，提升自己的Hadoop技能。在学习过程中，结合书中的理论知识和代码实现，将使你对Hadoop有更深入的...
通过分析Hadoop源代码，开发者可以更好地理解其内部工作原理，从而优化应用程序性能，调试问题，甚至为Hadoop贡献新的功能和改进。这个压缩包中的"完整版.pdf"文档很可能是对以上知识点的详细解读，值得深入学习和...
在深入分析Hadoop源代码之前，我们先了解一下Hadoop的核心组件和它们的作用。Hadoop是一个开源的分布式计算框架，由Apache基金会开发，主要用于处理和存储海量数据。它主要由两个核心部分组成：HDFS（Hadoop ...
7. **编程接口**：Hadoop提供了Java API来编写MapReduce程序，但也有如Hadoop Streaming这样的接口，允许使用其他语言（如Python、Perl）编写Mapper和Reducer。 8. **数据处理范式**：MapReduce遵循“批处理”处理...
文件"shizhan_03_hadoop"可能包含这个流量统计程序的源代码、配置文件或测试数据，对于深入理解其工作原理和使用方法至关重要。通过阅读和分析这些文件，我们可以更好地掌握如何在实际环境中应用Hadoop进行流量统计...