MongoDB集群與LBS應用系列(二)--與Hadoop集成

長期以來，我每開個系列，只有興趣寫一篇，很難持之與恆。為了克服這個長久以來的性格弱點，以及梳理工作半年的積累。最近一個月會寫兩篇關於Mongo在地理大數據方面的實踐和應用，一篇關於推薦系統的初期准備過程，一篇用戶行為矩陣的可視化。希望能夠立言為證，自我監督。

1.驅動准備

言歸正傳，前文MongoDB集群部署完畢之后，CRUD就是主要需求。NoSQL與普通關系數據庫不同的是，避免采用ORM框架對數據庫做操作，這樣會帶來明顯的性能下降[1]。使用原生的Driver是一個較為合理的選擇，Mongo支持的語言非常多，包括JS,Java,C,C++,Python,Scala等[2]。

如果是單純的MongoDB項目，我們會用NodeJS Driver，方便快捷，示例規范，值得推薦。在本文我使用Java Driver，主要是集成Hadoop工程方便。同時還會用到Mongo Hadoop Adapter 可以選擇到Github 下載源碼編譯，或者直接根據自己Hadoop集群版本選擇下載Jar包，添加到Hadoop安裝目錄的lib文件夾下[3]。但是在不少公有雲平台上，普通用戶是沒有修改Hadoop系統的權限，無法添加Jar包，所以在本文的示例代碼中，采用分布式緩存的方法添加這兩個Jar包。

2.實現原理與過程

其實Hadoop和MongoDB的集成，很大程度上是將Mongo作為Hadoop的輸入和輸出源，而Mongo Hadoop Adapter也是主要實現了BSONWritable，MongoInputformat等這些類，也就是說需要自定義Hadoop的序列化類以及輸入輸出格式。

2.1 Hadoop序列化與反序列化

序列化(serialization)將結構化對象轉化為二進制字節流，以便網絡傳輸和寫入磁盤。反序列化(deserialization)則是它的逆過程，將字節流轉化為結構化對象。分布式系統通常在進程通訊和持久化時候會使用序列化。Hadoop系統節點進程通信使用RPC，該協議存活時間非常短，因此需要其序列化格式具備以下特點：緊湊、快速、可擴展等。Hadoop提供了Writable接口，它定義了對數據的IO流，即需要實現readFields 和 Write兩個方法[4]。

2.2 Mongo Adapter的源碼實現

Mongo Hadoop Adater所實現的BSONWritable等類,源碼實現體現了上述的規范：

//輸出
public void write( DataOutput out ) throws IOException{
        BSONEncoder enc = new BasicBSONEncoder();
        BasicOutputBuffer buf = new BasicOutputBuffer();
        enc.set( buf );
        …………
    }
//輸入
public void readFields( DataInput in ) throws IOException{
        BSONDecoder dec = new BasicBSONDecoder();
        BSONCallback cb = new BasicBSONCallback();
        // Read the BSON length from the start of the record
       //字節流長度
        byte[] l = new byte[4];
        try {
            in.readFully( l );
            …………
            byte[] data = new byte[dataLen + 4];
            System.arraycopy( l, 0, data, 0, 4 );
            in.readFully( data, 4, dataLen - 4 );
            dec.decode( data, cb );
            _doc = (BSONObject) cb.get();
           ………………
    }

 

因此我們在編寫MapReduce程序的時候可以傳遞BsonWritable的key,value鍵值對，而Mongo構建於Bson之上，也就是說可以將MongoDB視為HDFS同性質的存儲節點即可。

3. 代碼實現

在Mongo-Hadoop網站有數個例子，但是講得不夠詳細，本文主要對它的金礦產量的例子做一個補充。完整的Hadoop項目一般包括Mapper,Reduceer,Job三個Java Class,以及一個一個配置文件(configuration.xml)來定義項目的輸入輸出等。Mongo-Hadoop項目會多一個mongo-defaults.xml，當然可以將兩者融合起來。

3.1  數據准備

從github中下載源碼包，它會包含examples/treasury_yield/src/main/resources/yield_historical_in.json文件，將該json文件上傳到Mongo所在的服務器，使用以下命令將它導入Mongo的testmr數據庫中的example collection中。

mongoimport --host 127.0.0.1 --port 27017 -d testmr -c example --file ./yield_historical_in.json

查看一下數據結構

use testmr
db.example.find().limit(1).pretty()

如下：

{
  "_id": ISODate("1990-01-25T19:00:00-0500"),
  "dayOfWeek": "FRIDAY", "bc3Year": 8.38,
  "bc10Year": 8.49,
  …
}

3.2  Mapper和Reducer還有Job以及mongo-defaults.xml

Mapper是從Mongo中讀取BSONObject

public class MongoTestMapper extends Mapper<Object,BSONObject, IntWritable, DoubleWritable>

以及處理讀過來的<key,value>鍵值對，並發到Reducer中匯總計算。注意value的類型。

public void map(final Object pkey, final BSONObject pvalue,final Context context)
        {
            final int year = ((Date)pvalue.get("_id")).getYear()+1990;
            double bdyear  = ((Number)pvalue.get("bc10Year")).doubleValue();
            try {
                context.write( new IntWritable( year ), new DoubleWritable( bdyear ));
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

Reducer會接受Mapper傳過來的鍵值對

public class MongoTestReducer extends Reducer<IntWritable,DoubleWritable,IntWritable,BSONWritable>

進行計算並將結果寫入MongoDB.請注意輸出的Value的類型是BSONWritable.

public void reduce( final IntWritable pKey,
            final Iterable<DoubleWritable> pValues,
            final Context pContext ) throws IOException, InterruptedException{
      int count = 0;
      double sum = 0.0;
      for ( final DoubleWritable value : pValues ){
          sum += value.get();
          count++;
      }

      final double avg = sum / count;

        BasicBSONObject out = new BasicBSONObject();
        out.put("avg", avg);
        pContext.write(pKey, new BSONWritable(out));
    }

 

Job作為MapReudce主類，主要使用DistributedCache分布式緩存來添加驅動包，並定義了任務的輸入配置等。如下所示：

//Using Distribute Cache,call it before job define.
        DistributedCache.createSymlink(conf);
//………………
//Using DistributedCache to add Driver Jar File
        DistributedCache.addFileToClassPath(new Path("/user/amap/data/mongo/mongo-2.10.1.jar"), conf);
        DistributedCache.addFileToClassPath(new Path("/user/amap/data/mongo/mongo-hadoop-core_cdh4.3.0-1.1.0.jar"), conf);

// job conf
        Job job = new Job(conf,"VentLam:Mongo-Test-Job");

mongo-defaults.xml 配置文件中定義了非常多的參數，我們只需要修改輸入輸出URI

   <!-- If you are reading from mongo, the URI -->
    <name>mongo.input.uri</name>
    <value>mongodb://127.0.0.1/testmr.example</value>
  </property>
  <property>
    <!-- If you are writing to mongo, the URI -->
    <name>mongo.output.uri</name>
    <value>mongodb://127.0.0.1/testmr.mongotest</value>
  </property>
  <property>

 

將整個java項目打包為名為mongotest的jar包，上傳到Hadoop集群，執行命令：

hadoop jar mongotest.jar org.ventlam.MongoTestJob

以后會將我的博客涉及到源碼都發布在https://github.com/ventlam/BlogDemo 中，這篇文章對應的是mongohadoop文件夾。

4.參考文獻

[1] What the overhead of Java ORM for MongoDB

http://stackoverflow.com/questions/10600162/what-the-overhead-of-java-orm-for-mongodb

[2] MongoDB Drivers and Client Libraries

http://docs.mongodb.org/ecosystem/drivers/

[3]Getting Started with Hadoop

http://docs.mongodb.org/ecosystem/tutorial/getting-started-with-hadoop/

[4] Interface Writable    http://hadoop.apache.org/docs/stable/api/

本作品由VentLam創作，采用知識共享署名-非商業性使用-相同方式共享 2.5 中國大陸許可協議進行許可。