MongoDB之Hadoop驱动介绍
1. 一些概念
Hadoop是一套Apache开源的分布式计算框架，其中包括了分布式文件系统DFS与分布式计算模型MapReduce，而MongoDB是一个面向文档的分布式数据库，它是NoSql的一种，而这里所要介绍的就是一个MongoDB的Hadoop驱动，这里就是把MongoDB作为MapReduce的输入源，充分利用MapReduce的优势来对MongoDB的数据进行处理与计算。2. MongoDB的Hadoop驱动
目前这个版本的Hadoop驱动还是测试版本，还不能应用到实际的生产环境中去。你可以从下面网址https://github.com/mongodb/mongo-hadoop下载到最新的驱动包，下面是它的一些依赖说明：
- 目前推荐用最新的Hadoop 0.20.203版本，或者是用Cloudera CHD3还做
- MongoDB的版本最好是用1.8+
- 还有是MongoDB的java驱动必须是2.5.3+
- 提供了一个Hadoop的Input和Output适配层，读于对数据的读入与写出
- 提供了大部分参数的可配置化，这些参数都可有XML配置文件来进行配置，你可以在配置文件中定义要查询的字段，查询条件，排序策略等
- 目前还不支持多Sharding的源数据读取
- 还不支持数据的split操作
3. 代码分析
运行其examples中的WordCount.java代码
- // 事先在MongoDB的test数据库的in表中加入的测试样本,使用如下方法
- /**
- * test.in db.in.insert( { x : "eliot was here" } ) db.in.insert( { x :
- * "eliot is here" } ) db.in.insert( { x : "who is here" } ) =
- */
- ublic class WordCount {
- private static final Log log = LogFactory.getLog( WordCount.class );
- // 这是一个Map操作
- public static class TokenizerMapper extends Mapper<Object, BSONObject, Text, IntWritable> {
- private final static IntWritable one = new IntWritable( 1 );
- private final Text word = new Text();
- public void map( Object key , BSONObject value , Context context ) throws IOException, InterruptedException{
- System.out.println( "key: " + key );
- System.out.println( "value: " + value );
- // 对词进行按空格切分
- final StringTokenizer itr = new StringTokenizer( value.get( "x" ).toString() );
- while ( itr.hasMoreTokens() ) {
- word.set( itr.nextToken() );
- context.write( word, one ); // 这里的key为词，而value为1
- }
- }
- }
- // 这是Reduce操作，用于计算词出现的频率
- public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
- private final IntWritable result = new IntWritable();
- public void reduce( Text key , Iterable<IntWritable> values , Context context ) throws IOException, InterruptedException{
- // 计算词出现的频率，把相同词的value相加
- int sum = 0;
- for ( final IntWritable val : values ) {
- sum += val.get();
- }
- result.set( sum );
- context.write( key, result ); // key为单个词,value为这个词所对应的词频率
- }
- }
- public static void main( String[] args ) throws Exception{
- final Configuration conf = new Configuration();
- // 定义MongoDB数据库的输入与输出表名，这里是调用本地的MongoDB，默认端口号为27017
- MongoConfigUtil.setInputURI( conf, "mongodb://localhost/test.in" );
- MongoConfigUtil.setOutputURI( conf, "mongodb://localhost/test.out" );
- System.out.println( "Conf: " + conf );
- final Job job = new Job( conf , "word count" );
- job.setJarByClass( WordCount.class );
- // 定义Mapper,Reduce与Combiner类
- job.setMapperClass( TokenizerMapper.class );
- job.setCombinerClass( IntSumReducer.class );
- job.setReducerClass( IntSumReducer.class );
- // 定义Mapper与Reduce的输出key/value的类型
- job.setOutputKeyClass( Text.class );
- job.setOutputValueClass( IntWritable.class );
- // 定义InputFormat与OutputFormat的类型
- job.setInputFormatClass( MongoInputFormat.class );
- job.setOutputFormatClass( MongoOutputFormat.class );
- System.exit( job.waitForCompletion( true ) ? 0 : 1 );
- }
4. 分块机制的简单介绍
这里没有实现对不同shard的split操作，也就是说，对于分布在不同shard上的数据，只会产生一个Map操作。
这里本人提供了一个分片的思路，有兴趣的可以讨论一下。
我们知道，对于Collection分块后，会产生一个Config数据库，在这个数据库下有一个叫做chunks的表，其中每个chunk记录了start_row与end_row，而这些chunk可以分布在不同的shard上，我们可以通过分析这个Collection来得到每个shard上的chunk信息，从而把每个shard上的chunk信息组合成一个InputSplit,这就是这里的MongoInputSplit,这样的话，只要去修改MongoInputFormat这个类的getSplits这个方法，加入对chunks表的分析，得到shard的信息，这样就可以实现多split的Map操作，对于不同的Shard,每个Map都会调用本地的Mongos代理服务，这样就实现了移动计算而不是移动数据的目的。
相关推荐
lbyd0 2020-11-17
BigYellow 2020-11-16
sushuanglei 2020-11-12
我心似明月 2020-11-09
zhushenghan 2020-11-09
sunnnyduan 2020-10-16
不要皱眉 2020-10-14
xiaohai 2020-09-29
songxiugongwang 2020-09-22
萌亖 2020-09-17
LuckyLXG 2020-09-08
sdmzhu 2020-09-01
mkhhxxttxs 2020-09-16
xiaohai 2020-09-16
newcome 2020-09-09
jaylong 2020-08-19
大秦铁骑 2020-08-19
thatway 2020-08-19