package cn.luxh.app; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; import org.apache.hadoop.hbase.mapreduce.TableMapper; import org.apache.hadoop.hbase.mapreduce.TableReducer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; /** * @author Luxh * */ public class WordStat { /** * TableMapper<Text,IntWritable> Text:输出的key类型,IntWritable:输出的value类型 */ public static class MyMapper extends TableMapper<Text,IntWritable>{ private static IntWritable one = new IntWritable(1); private static Text word = new Text(); @Override protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException { //表里面只有一个列族,所以我就直接获取每一行的值 String words = Bytes.toString(value.list().get(0).getValue()); StringTokenizer st = new StringTokenizer(words); while (st.hasMoreTokens()) { String s = st.nextToken(); word.set(s); context.write(word, one); } } } /** * TableReducer<Text,IntWritable> Text:输入的key类型,IntWritable:输入的value类型,ImmutableBytesWritable:输出类型 */ public static class MyReducer extends TableReducer<Text,IntWritable,ImmutableBytesWritable>{ @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for(IntWritable val:values) { sum+=val.get(); } //添加一行记录,每一个单词作为行键 Put put = new Put(Bytes.toBytes(key.toString())); //在列族result中添加一个标识符num,赋值为每个单词出现的次数 //String.valueOf(sum)先将数字转化为字符串,否则存到数据库后会变成\x00\x00\x00\x这种形式 //然后再转二进制存到hbase。 put.add(Bytes.toBytes("result"), Bytes.toBytes("num"), Bytes.toBytes(String.valueOf(sum))); context.write(new ImmutableBytesWritable(Bytes.toBytes(key.toString())),put); } } public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = HBaseConfiguration.create(); Job job = new Job(conf,"wordstat"); job.setJarByClass(Blog.class); Scan scan = new Scan(); //指定要查询的列族 scan.addColumn(Bytes.toBytes("content"),null); //指定Mapper读取的表为word TableMapReduceUtil.initTableMapperJob("word", scan, MyMapper.class, Text.class, IntWritable.class, job); //指定Reducer写入的表为stat TableMapReduceUtil.initTableReducerJob("stat", MyReducer.class, job); System.exit(job.waitForCompletion(true)?0:1); } }
相关推荐
MapReduce是一种分布式计算模型,常用于处理大规模数据集,而HBase是一个分布式、列式存储的NoSQL数据库,适合处理大规模结构化数据。通过MapReduce,我们可以高效地在不同HBase表之间移动大量数据。 首先,我们...
- **概念**:MapReduce是一种编程模型,用于处理并生成大数据集。 - **流程**: - **Map阶段**:将输入数据分割成小块,然后对每个数据块执行映射操作。 - **Reduce阶段**:将映射后的结果汇总,进行进一步处理。 ...
13. 使用 MapReduce 实现 Join 操作:使用 MapReduce 来实现数据的 Join 操作,以便将多个数据源合并成一个结果。 14. 使用 MapReduce 实现排序:使用 MapReduce 来实现数据的排序,以便对数据进行排序处理。 15. ...
- MapReduce:作为一种编程模型,用于并行处理大数据集。 - Avro:一个用于序列化数据的远程过程调用(RPC)和持久化数据存储的框架。 - Chukwa和Ganglia:用于集群监控的工具。 围绕Hadoop的技术公司 Hadoop的...
MapReduce是一种分布式计算模型,由Google在2004年提出,主要用于处理和生成大规模数据集。它将复杂的并行计算任务分解为两个主要阶段:Map(映射)和Reduce(化简),使得大规模数据处理变得简单易懂。C语言实现...
4. **输出(Output)**:完成化简操作后,结果会被写入到HDFS(Hadoop Distributed File System)中,作为新的数据源供后续任务使用。 本书涵盖了多个关键知识点,包括但不限于: 1. **Hadoop环境搭建**:讲解如何...
在实际操作中,MapReduce作业的输入可以是HDFS、HBase等Hadoop内的数据源,甚至可以是外部数据库或其他非Hadoop存储系统中的数据。作业的输出则可以根据需求写回到HDFS或其他存储系统,供后续分析或应用使用。 总的...
在压缩包中的代码文件,可能是各个章节的示例程序,涵盖了Hadoop的安装配置、数据输入输出、MapReduce编程模型、Hadoop流(Hadoop Streaming)、Hive数据仓库、Pig脚本语言、HBase分布式数据库、Oozie工作流管理、...
这个教程涵盖了多个关键领域,包括HBase数据库、MapReduce编程模型、多语言MapReduce实现、Chukwa监控系统、Greenplum架构以及Flume日志收集系统。下面,我们将详细探讨这些知识点。 首先,HBase是一个分布式、面向...
5. **数据输入与输出**:书中详细讲解了Hadoop如何处理各种类型的数据源,如文本文件、Avro数据、Parquet列式存储格式,以及如何通过不同的OutputFormat输出结果。 6. **MapReduce编程**:通过实例展示了如何编写...
通过这本书,读者应能够掌握如何安装和配置Hadoop集群,如何使用MapReduce编程模型进行数据处理,以及如何利用Hadoop生态系统中的工具解决实际问题。此外,读者还能够学习到如何优化和调整Hadoop集群的性能,以适应...
第四章《MapReduce入门编程》深入MapReduce编程模型,解析Map和Reduce阶段的工作原理。这一章通过实例演示如何编写Java MapReduce程序,讲解输入输出格式、分区器、Combiner等相关概念,帮助理解MapReduce的执行流程...
这个模型与传统的MapReduce编程模型有所不同,它更加灵活和强大。 对于实时流数据处理,Apache Beam提供了一套非常有用的工具和概念。它能够处理实时数据流,并且可以在数据流上设置窗口,实现对数据的实时分析和...
3. **MapReduce编程模型**:MapReduce是Hadoop处理大规模数据的核心计算模型,由“Map”和“Reduce”两个阶段组成。Map将输入数据分割并转换,Reduce则对Map的输出进行聚合和总结。 4. **YARN**:YARN是Hadoop的...
3. **Chap 3 - MapReduce编程**:涵盖了MapReduce的编程模型,包括Mapper和Reducer类的实现,以及Combiner和Partitioner的作用。Combiner可以局部聚合数据,减少网络传输;Partitioner则决定数据在Reducer间的分布。...
MapReduce则是一种编程模型,用于大规模数据集的并行计算。 在Hadoop源代码中,我们可以深入理解其内部工作原理,包括数据块的分发、容错机制、任务调度和数据本地化等关键功能。以下是Hadoop源代码中的一些关键...
MapReduce则是一种编程模型,用于大规模数据集的并行计算。 二、Hadoop的应用范围 Hadoop广泛应用于数据挖掘、日志分析、互联网搜索、推荐系统等领域。通过其分布式处理能力,可以处理PB级别的数据,帮助企业快速...