Using SQL to query MongoDB with HIVE

This project aims to show how to use the MongoDB connector for Hadoop in order to make SQL queries on data stored in MongoDB.

Design

This is based on Hive : this tool needs Hadoop to work. The application works in a standalone mode with an embedded Hadoop cluster.

The first step is to use the Mini Cluster provided by Hadoop : it is a simple hadoop cluster where all the components are running in a single JVM (HDFS : namenode, datanode; YARN : resource manager, node manager).
The second step is to start the HIVE components: a metastore and a server.

All these things can be easily done with Spring for Apache Hadoop.

In this example, there are two modes:

Shell mode : you can write your queries in an interactive shell
Server mode : you can start the server part (Hadoop, Hive) and a client application will connect to this server using the JDBC driver

Configuration

See config files in src/main/resources.

hadoop-hive.xml : Hadoop cluster configuration
hive-shell.xml : Configuration of the interactive shell
jdbc-client.xml : JDBC configuration

Compilation

    mvn clean compile

Run

Start MongoDB

Start the Hadoop part :

Option 1 : start the Hive shell (it includes an embedded Hadoop cluster) using hiveshell.sh

echo "Start the hive shell with the following arguments : $*"
rm -f hadoop_mongodb.log
mvn exec:exec -Dexec.arguments="$*" -Phiveshell

Start the hive shell with the following arguments :
[INFO] Scanning for projects...
[INFO]
[INFO] ------------------------------------------------------------------------
[INFO] Building hadoop-mongodb 0.0.1-SNAPSHOT
[INFO] ------------------------------------------------------------------------
[INFO]
[INFO] --- exec-maven-plugin:1.3.2:exec (default-cli) @ hadoop-mongodb ---
Formatting using clusterid: testClusterID
+---------------------------+
| Welcome to the Hive Shell |
+---------------------------+
>
>

Now, you can query your MongoDB (see Queries chapter) !

Option 2 : start the Hadoop mini-cluster and a JDBC client

mvn exec:exec -Dexec.arguments="$*" -Pcluster

mvn exec:exec -Dexec.arguments="$*" -Pjdbcclient

Queries

Create tables :

>  create table jobs (id string, name string, job string) 
   stored by 'com.mongodb.hadoop.hive.MongoStorageHandler' 
   WITH SERDEPROPERTIES('mongo.columns.mapping'='{"id":"_id"}') 
   TBLPROPERTIES('mongo.uri'='mongodb://localhost:27017/test.jobs');

OK

> create table individuals
  (id struct<oid:string, bsontype:int>, name string, age int, 
   address struct<zipcode:int, city:string>)
  stored by 'com.mongodb.hadoop.hive.MongoStorageHandler' 
  WITH SERDEPROPERTIES('mongo.columns.mapping'='{"id":"_id"}') 
  TBLPROPERTIES('mongo.uri'='mongodb://localhost:27017/test.individuals', 'columns.comments'='');

OK

Get table description:

 > describe extended individuals;
OK
id                      struct<oid:string,bsontype:int> from deserializer
name                    string                  from deserializer
age                     int                     from deserializer
address                 struct<zipcode:int,city:string> from deserializer

Detailed Table Information      Table(tableName:individuals, dbName:default, owner:cdnhdckk, createTime:1409238198, lastAccessTime:0, retention:0, sd:StorageDescriptor(cols:[FieldSchema(name:id, type:struct<oid:string,bsontype:int>, comment:null), FieldSchema(name:name, type:string, comment:null), FieldSchema(name:age, type:int, comment:null), FieldSchema(name:address, type:struct<zipcode:int,city:string>, comment:null)], location:file:/tmp/individuals, inputFormat:com.mongodb.hadoop.hive.input.HiveMongoInputFormat, outputFormat:com.mongodb.hadoop.hive.output.HiveMongoOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:com.mongodb.hadoop.hive.BSONSerDe, parameters:{serialization.format=1, mongo.columns.mapping={"id":"_id"}}), bucketCols:[], sortCols:[], parameters:{}, skewedInfo:SkewedInfo(skewedColNames:[], skewedColValues:[], skewedColValueLocationMaps:{}), storedAsSubDirectories:false), partitionKeys:[], parameters:{numFiles=0, mongo.uri=mongodb://localhost:27017/test.individuals, columns.comments=, transient_lastDdlTime=1409238198, COLUMN_STATS_ACCURATE=false, totalSize=0, numRows=-1, storage_handler=com.mongodb.hadoop.hive.MongoStorageHandler, rawDataSize=-1}, viewOriginalText:null, viewExpandedText:null, tableType:MANAGED_TABLE)

Basic "select":

> select * from individuals;
OK
{"oid":"53f4a478e4b0db0f5dbba307","bsontype":8} Bob     34      {"zipcode":12345,"city":"ICI"}
{"oid":"53f4a478e4b0db0f5dbba308","bsontype":8} Bobette 33      {"zipcode":54321,"city":"LA-BAS"}

Count : Hive builds a Map/Reduce job for this kind of query

 > select count(*) from individuals;
Total jobs = 1
Launching Job 1 out of 1
Number of reduce tasks determined at compile time: 1
In order to change the average load for a reducer (in bytes):
  set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
  set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
  set mapreduce.job.reduces=<number>
Starting Job = job_1409311417015_0001, Tracking URL = http://localhost:27800/proxy/application_1409311417015_0001/
Kill Command = /usr/bin/hadoop job  -kill job_1409311417015_0001
Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 1
2014-08-29 13:28:54,017 Stage-1 map = 0%,  reduce = 0%
2014-08-29 13:29:01,791 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 1.78 sec
2014-08-29 13:29:10,283 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 3.54 sec
MapReduce Total cumulative CPU time: 3 seconds 540 msec
Ended Job = job_1409311417015_0001
MapReduce Jobs Launched:
Job 0: Map: 1  Reduce: 1   Cumulative CPU: 3.54 sec   HDFS Read: 363 HDFS Write: 2 SUCCESS
Total MapReduce CPU Time Spent: 3 seconds 540 msec
OK
2

Join : in this case, Hive also builds a Map/Reduce job

 > select i.name, j.job from individuals i, jobs j where i.name=j.name;
Total jobs = 1
Launching Job 1 out of 1
Number of reduce tasks is set to 0 since there's no reduce operator
Starting Job = job_1409311417015_0002, Tracking URL = http://localhost:27800/proxy/application_1409311417015_0002/
Kill Command = /usr/bin/hadoop job  -kill job_1409311417015_0002
Hadoop job information for Stage-3: number of mappers: 1; number of reducers: 0
2014-08-29 13:32:33,709 Stage-3 map = 0%,  reduce = 0%
2014-08-29 13:32:42,354 Stage-3 map = 100%,  reduce = 0%, Cumulative CPU 2.71 sec
MapReduce Total cumulative CPU time: 2 seconds 710 msec
Ended Job = job_1409311417015_0002
MapReduce Jobs Launched:
Job 0: Map: 1   Cumulative CPU: 2.71 sec   HDFS Read: 349 HDFS Write: 39 SUCCESS
Total MapReduce CPU Time Spent: 2 seconds 710 msec
OK
Bob     CTO
Bobette UNKNOWN
Bobette DRIVER

Load data : in order to load data in non-native table, you have to create an temporary table

First step : create a temporary table

create table src_individuals (name string, age int, zipcode int, city string)
   row format delimited fields terminated by ',';

Second step : load the data in the temporary table

> load data local inpath 'file:///tmp/data.csv' into table src_individuals;
Copying data from file:/tmp/data.csv
Copying file: file:/tmp/data.csv
Loading data to table default.src_individuals
Table default.src_individuals stats: [numFiles=2, numRows=0, totalSize=82, rawDataSize=0]
OK

Last step : insert the data in the collection (it uses a User-Defined Function in order to create an ObjectID instance). Again, for the insert, Hive generates a Map/Reduce job.

> create temporary function newObjectId as 'hadoopmongo.hive.UDFObjectId';
OK

> insert into table individuals select newObjectId() as id, name, age, named_struct('zipcode', zipcode, 'city', city) as address from src_individuals;
Total jobs = 1
Launching Job 1 out of 1
Number of reduce tasks is set to 0 since there's no reduce operator
Starting Job = job_1409311417015_0003, Tracking URL = http://localhost:27800/proxy/application_1409311417015_0003/
Kill Command = /usr/bin/hadoop job  -kill job_1409311417015_0003
Hadoop job information for Stage-0: number of mappers: 1; number of reducers: 0
2014-08-29 13:44:30,593 Stage-0 map = 0%,  reduce = 0%
2014-08-29 13:44:39,147 Stage-0 map = 100%,  reduce = 0%, Cumulative CPU 1.91 sec
MapReduce Total cumulative CPU time: 1 seconds 910 msec
Ended Job = job_1409311417015_0003
MapReduce Jobs Launched:
Job 0: Map: 1   Cumulative CPU: 1.91 sec   HDFS Read: 246 HDFS Write: 0 SUCCESS
Total MapReduce CPU Time Spent: 1 seconds 910 msec
OK

albertleecn/hadoop-mongodb

Using SQL to query MongoDB with HIVE

Design

Configuration

Compilation

Run

Queries