Using MongoDB with Hadoop & Spark: Part 2 - Hive Example

< View all blog posts
Matt Kalan
February 17, 2015
Category: Technical

Update: August 4th 2016 Since this original post, MongoDB has released a new certified connector for Spark. Click through for a tutorial on using the new MongoDB Connector for Apache Spark.

Welcome to part two of our three-part series on MongoDB and Hadoop. In part one, we introduced Hadoop and how to set it up. In this post, we'll look at a Hive example.

  1. Introduction & Setup of Hadoop and MongoDB
  2. Hive Example
  3. Spark Example & Key Takeaways

For more detail on the use case, see the first paragraph of part 1.

Summary

Use case: aggregating 1 minute intervals of stock prices into 5 minute intervals
Input: 1 minute stock prices intervals in a MongoDB database
Simple Analysis: performed in:

  • Hive
  • Spark

Output: 5 minute stock prices intervals in Hadoop

Hive Example

I ran the following example from the Hive command line (simply typing the command “hive” with no parameters), not Cloudera’s Hue editor, as that would have needed additional installation steps. I immediately noticed the criticism people have with Hive, that everything is compiled into MapReduce which takes considerable time. I ran most things with just 20 records to make the queries run quickly.

This creates the definition of the table in Hive that matches the structure of the data in MongoDB. MongoDB has a dynamic schema for variable data shapes but Hive and SQL need a schema definition.

You can also find the below Hive SQL statements in hive-ohlcbars-example.sql in my Github repo.

CREATE EXTERNAL TABLE minute_bars
(

id STRUCT,
      Symbol STRING,
      Timestamp STRING,
      Day INT,
      Open DOUBLE,
      High DOUBLE,
      Low DOUBLE,
      Close DOUBLE,
      Volume INT
)
STORED BY 'com.mongodb.hadoop.hive.MongoStorageHandler'
WITH SERDEPROPERTIES('mongo.columns.mapping'='{"id":"_id",
 "Symbol":"Symbol", "Timestamp":"Timestamp", "Day":"Day", "Open":"Open", "High":"High", "Low":"Low", "Close":"Close", "Volume":"Volume"}')
TBLPROPERTIES('mongo.uri'='mongodb://localhost:27017/marketdata.minbars');

Recent changes in the Apache Hive repo make the mappings necessary even if you are keeping the field names the same. This should be changed in the MongoDB Hadoop Connector soon if not already by the time you read this.

Then I ran the following command to create a Hive table for the 5 minute bars:

CREATE TABLE five_minute_bars
(

   id STRUCT,
    Symbol STRING,
    Timestamp STRING,
    Open DOUBLE,
     High DOUBLE,
     Low DOUBLE,
    Close DOUBLE
);

This insert statement uses the SQL windowing functions to group 5 1-minute periods and determine the OHLC for the 5 minutes. There are definitely other ways to do this but here is one I figured out. Grouping in SQL is a little different from grouping in the MongoDB aggregation framework (in which you can pull the first and last of a group easily), so it took me a little while to remember how to do it with a subquery.

The subquery takes each group of 5 1-minute records/documents, sorts them by time, and takes the open, high, low, and close price up to that record in each 5-minute period. Then the outside WHERE clause selects the last 1-minute bar in that period (because that row in the subquery has the correct OHLC information for its 5-minute period). I definitely welcome easier queries to understand but you can run the subquery by itself to see what it’s doing too.

INSERT INTO TABLE five_minute_bars
SELECT m.id, m.Symbol, m.OpenTime as Timestamp, m.Open, m.High, m.Low, m.Close
FROM
(SELECT

     id,
      Symbol,
      FIRST_VALUE(Timestamp)
      OVER (
            PARTITION BY floor(unix_timestamp(Timestamp, 'yyyy-MM-dd HH:mm')/(5*60))
             ORDER BY Timestamp)

     as OpenTime,

     LAST_VALUE(Timestamp)
      OVER (
            PARTITION BY floor(unix_timestamp(Timestamp, 'yyyy-MM-dd HH:mm')/(5*60))
             ORDER BY Timestamp)

     as CloseTime,

     FIRST_VALUE(Open)
      OVER (
            PARTITION BY floor(unix_timestamp(Timestamp, 'yyyy-MM-dd HH:mm')/(5*60))
             ORDER BY Timestamp)

     as Open,
       MAX(High)
      OVER (
            PARTITION BY floor(unix_timestamp(Timestamp, 'yyyy-MM-dd HH:mm')/(5*60))
             ORDER BY Timestamp)

     as High,

     MIN(Low)
      OVER (
            PARTITION BY floor(unix_timestamp(Timestamp, 'yyyy-MM-dd HH:mm')/(5*60))
             ORDER BY Timestamp)

     as Low,
      LAST_VALUE(Close)
      OVER (
            PARTITION BY floor(unix_timestamp(Timestamp, 'yyyy-MM-dd HH:mm')/(5*60))
             ORDER BY Timestamp)

     as Close
FROM minute_bars)
as m
WHERE unix_timestamp(m.CloseTime, 'yyyy-MM-dd HH:mm') - unix_timestamp(m.OpenTime, 'yyyy-MM-dd HH:mm') = 60*4;

I can definitely see the benefit of being able to use SQL to access data in MongoDB and optionally in other databases and file formats, all with the same commands, while the mapping differences are handled in the table declarations. The downside is that the latency is quite high, but that could be made up some with the ability to scale horizontally across many nodes. I think this is the appeal of Hive for most people - they can scale to very large data volumes using traditional SQL, and latency is not a primary concern.

Post #3 in this blog series shows similar examples using Spark.

  1. Introduction & Setup of Hadoop and MongoDB
  2. Hive Example
  3. Spark Example & Key Takeaways


Want to learn more about Apache Spark and MongoDB? Read our white paper on turning analytics into real-time action.

Read more about Apache Spark and MongoDB


<< Read Part 1

Read Part 3 >>



About Matt Kalan

Matt Kalan is a Sr. Solution Architect at MongoDB, with extensive experience helping more than 300 customers in financial services and other industries solve business problems with technology. Before MongoDB, Matt grew Progress Software’s Apama Algorithmic Trading and Complex Event Processing (CEP) Platform business in North America and later sold broader operational intelligence solutions to FS firms. He previously worked for Caplin Systems selling solutions to stream real-time market data over the web to FX and FI portals, and for Sapient providing consulting services to global 2000 clients.

comments powered by Disqus