by Ori Stitelman

In this case study we examine one of many approaches our data science team, here at m6d, takes toward synthesizing the immense amount of data that we are able to extract using Hive. m6d is a display advertising prospecting company. Our role is to create machine learning algorithms that are specifically tailored toward finding the best new prospects for an advertising campaign. These algorithms are layered on top of a delivery engine that is tied directly into a myriad of real time bidding exchanges that provide a means to purchase locations on websites to display banner advertisements on behalf of our clients. The m6d display advertising engine is involved in billions of auctions a day and tens of millions of advertisements daily. Naturally, such a system produces an immense amount of data. A large portion of the records that are generated by our company’s display advertising delivery system are housed in m6d’s Hadoop cluster and, as a result, Hive is the primary tool our data science team uses to interact with the these logs.

Hive gives our data science team a way to extract and manipulate large amounts of data. In fact, it allows us to extract samples and summarize data that prior to using Hive could not be analyzed as efficiently, or at all, because of the immense size. Despite the fact that Hive allows us access to huge amounts of data at rates many times faster than before, it does not change the fact that most of the tools that we were previously familiar with as data scientists are not always able to analyze data samples of the size we can now produce. In summary, Hive provides us a great tool to extract huge amounts of data; however, the toolbox of data science, or statistical learning, methods that we as data scientists are used to using cannot easily accommodate the new larger data sets without substantial changes.

Many different software packages have been developed or are under development for both supervised and unsupervised learning on large data sets. Some of these software packages are stand alone software implementations, such as Vowpal Wabbit and BBR, while others are implementations within a larger infrastructure such as Mahout for Hadoop or the multitude of “large data” packages for R. A portion of these algorithms take advantage of parallel programing approaches while others rely on different methods to achieve scalability.

The primary tool for statistical learning for several of the data scientists in our team is R. It provides a large array of packages that are able to perform many statistical learning methods. More importantly, we have a lot of experience with it, know how its packages perform, understand their features, and are very familiar with its documentation. However, one major drawback of R is that by default it loads the entire data set into memory. This is a major limitation considering that the majority of the data sets that we extract from Hive and are able to analyze today are much larger than what can fit in memory. Moreover, once the data in R is larger than what is able to fit in memory, the system will start swapping, which leads to the system thrashing and massive decreases in processing speed.^[26]

In no way are we advocating ignoring the new tools that are available. Obviously, it is important to take advantage of the best of these scalable technologies, but only so much time can be spent investigating and testing new technology. So now we are left with a choice of either using the new tools that are available for large data sets or downsampling our data to fit into the tools that we are more familiar with. If we decide to use the new tools, we can gain signal by letting our data learn off of more data, and as a result the variance in our estimates will decrease. This is particularly appealing in situations where the outcome is very rare. However, learning these new tools takes time and there is an opportunity cost of using that time to learn new tools rather than answering other questions that have particular value to the company.

Alternatively, we can downsample the data to obtain something that can fit in the old tools we have at our disposal, but must deal with a loss of signal and increased variance in our estimates. However, this allows us to deal with tools with which we are familiar and the features that they provide. Thus, we are able to retain the functionality of our current toolbox at the price of losing some signal. However, these are not the only two possible approaches. In this case study, we highlight a way that we can both retain the functionality of the current toolbox as well as gain signal, or decrease variance, by using a larger sample, or all, of the data available to us.

Figure 23-1 shows the probability of converting versus the score from an algorithm designed to rank prospects for an advertising campaign. Higher scores should indicate a higher probability of conversion. This plot clearly shows that the top individuals are converting at a lower rate than some of the lower scoring browsers. That is, browsers with scores greater than 1 convert at a lower rate than individuals with scores between 0.5 and 1.0. Considering that some campaigns only target a very small percentage of the overall population, it is important the best prospects are among the top scorers.

Figure 23-1. Probability of conversion versus score

The line that expresses the relationship between the score and the conversion probability seen in Figure 23-1 is generated using a generalized additive model (GAM) in the statistical programming package R.^[27] The details about GAMs will not be presented here. For the purpose of this case study it can be thought of as a black box that produces for each score a prediction of the conversion rate. The browsers can then be re-ranked based on the predicted conversion rate; thus, the predicted conversion rate becomes the new score.

The new ranking can be generated in the following way. First, extract the scores for each individual browser and then follow them for some designated period of time, say five days, and record if they took the desired action, and thus converted. Consider a Hive table called scoretable that has the following information and is partitioned on date and subpartitioned by offer.

The following query can then be used to extract a set of data from scoretable for use in R to estimate the GAM line that predicts conversion for different levels of score like in the preceding table:

SELECT score,convert
FROM scoretable
WHERE date >= (…)  AND date <= (…)
AND offer = (…);
1.2347 0
3.2322 1
0.0013 0
0.3441 0

This data is then loaded into R and the following code is used to create the predicted conversion probability versus score, as in the preceding table:

library(mgcv)
g1=gam(convert~s(score),family=binomial,data=[data frame name])

The issue with this approach is that it only can be used for a limited number of days of data because the data set gets too large and R begins thrashing for any more than three days of data. Moreover, it takes approximately 10 minutes of time for each campaign to do this for about three days of data. So, running this analysis for about 300 campaigns for a single scoring method took about 50 hours for three days of data.

By simply extracting the data from Hive in a slightly different way and making use of the feature of the gam function in mgcv that allows for frequency weights, the same analysis may be done using more data, and thus gaining more signal, at a much faster rate. This is done by selecting the data from Hive by rounding the score to the nearest hundredth and getting frequency weights for each rounded score, convert combination by using a GROUP BY query. This is a very common approach for dealing with large data sets and in the case of these scores there should be no loss of signal due to rounding because there is no reason to believe that individuals with scores that differ by less than 0.001 are any different from each other. The following query would select such a data set:

SELECT round(score,2) as score,convert,count(1) AS freq
        FROM scoretable
        WHERE date >= [start.date] and date <= [end.date] and offer = [chosen.offer]
        GROUP BY round(score,2),convert;
1.23 0 500
3.23 1 22
0.00 0 127
0.34 0 36

The resulting data set is significantly smaller than the original approach presented that does not use frequency weights. In fact, the initial data set for each offer consisted of millions of records, and this new data set consists of approximately 6,500 rows per offer. The new data is then loaded into R and the following code may be used to generate the new GAM results:

library(mgcv)
g2=gam(convert~s(score),family=binomial,weights=freq,
 data=[frequency weight data frame name])

(We wrapped the line.)

The previously presented approach took 10 minutes per offer to create the GAM for only three days of data, compared to the frequency-weighted approach which was able to create the GAM based on seven days of data in approximately 10 seconds. Thus, by using frequency weights, the analysis for the 300 campaigns was able to be done in 50 minutes compared to 50 hours using the originally presented approach. This increase in speed was also realized while using more than twice the amount of data resulting in more precise estimates of the predicted conversion probabilities. In summary, the frequency weights allowed for a more precise estimate of the GAM in significantly less time.

In the presented case study, we showed how by rounding the continuous variables and grouping like variables with frequency weights, we were both able to get more precise estimates by using more data and fewer computational resources, resulting in quicker estimates. The example shown was for a model with a single feature, score. In general, this is an approach that will work well for a low number of features or a larger number of sparse features. The above approach may be extended to higher dimensional problems as well using some other small tricks. One way this can be done for a larger number of variables is by bucketing the variables, or features, into binary variables and then using GROUP BY queries and frequency weights for those features. However, as the number of features increases, and those features are not sparse, there is little value gained by such an approach and other alternative methods must be explored, or software designed for larger data sets must be embraced.

by David Ha and Rumit Patel

Sorting data and identifying the top N elements is straightforward. You order the whole data set by some criteria and limit the result set to N. But there are times when you need to group like elements together and find the top N elements within that group only. For example, identifying the top 10 requested songs for each recording artist or the top 100 best-selling items per product category and country. Several database platforms define a rank() function that can support these scenarios, but until Hive provides an implementation, we can create a user-defined function to produce the results we want. We will call this function p_rank() for psuedorank, leaving the name rank() for the Hive implementation.

Say we have the following product sales data and we want to see the top three items per category and country:

In most SQL systems:

SELECT
 category,country,product,sales,rank
FROM (
 SELECT
   category,country,product, sales,
   rank() over (PARTITION BY category, country ORDER BY sales DESC) rank
 FROM p_rank_demo) t
WHERE rank <= 3

To achieve the same result using HiveQL, the first step is partitioning the data into groups, which we can achieve using the DISTRIBUTE BY clause. We must ensure that all rows with the same category and country are sent to the same reducer:

DISTRIBUTE BY
 category,
 country

The next step is ordering the data in each group by descending sales using the SORT BY clause. While ORDER BY effects a total ordering across all data, SORT BY affects the ordering of data on a specific reducer. You must repeat the partition columns named in the DISTRIBUTE BY clause:

SORT BY
 category,
 country,
 sales DESC

Putting everything together, we have:

ADD JAR p-rank-demo.jar;
CREATE TEMPORARY FUNCTION p_rank AS 'demo.PsuedoRank';

SELECT
 category,country,product,sales,rank
FROM (
 SELECT
   category,country,product,sales,
   p_rank(category, country) rank
 FROM (
   SELECT
     category,country,product,
     sales
   FROM p_rank_demo
   DISTRIBUTE BY
     category,country
   SORT BY
     category,country,sales desc) t1) t2
WHERE rank <= 3

The subquery t1 organizes the data so that all data belonging to the same category and country are sorted by descending sales count. The next query t2 then uses p_rank() to assign a rank to each row within the group. The outermost query filters the rank to be in the top three:

p_rank() is implemented as a generic UDF whose parameters are all the identifying attributes of the group, which, in this case, are category and country. The function remembers the previous arguments, and so long as the successive arguments match, it increments and returns the rank. Whenever the arguments do not match, the function resets the rank back to 1 and starts over.

This is just one simple example of how p_rank() can be used. You can also find the 10th to 15th bestsellers by category and country. Or, if you precalculate the counts of products in each category and country, you can use p_rank() to calculate percentiles using a join. For example, if there were 1,000 products in the “movies” and “us” group, the 50th, 70th, and 95th quantiles would have rank 500, 700, and 950, respectively. Please know that p_rank() is not a direct substitute for rank() because there will be differences in some circumstances. rank() returns the same value when there are ties, but p_rank() will keep incrementing, so plan accordingly and test with your data.

Lastly, here is the implementation. It is public domain so feel free to use, improve, and modify it to suit your needs:

package demo;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.
  PrimitiveObjectInspectorFactory;

public class PsuedoRank extends GenericUDF {
 /**
  * The rank within the group. Resets whenever the group changes.
  */
 private long rank;

 /**
  * Key of the group that we are ranking. Use the string form
  * of the objects since deferred object and equals do not work
  * as expected even for equivalent values.
  */
 private String[] groupKey;

 @Override
 public ObjectInspector initialize(ObjectInspector[] oi)
     throws UDFArgumentException {
   return PrimitiveObjectInspectorFactory.javaLongObjectInspector;
 }

 @Override
 public Object evaluate(DeferredObject[] currentKey) throws HiveException {
   if (!sameAsPreviousKey(currentKey)) {
     rank = 1;
   }
   return new Long(rank++);
 }

 /**
  * Returns true if the current key and the previous keys are the same.
  * If the keys are not the same, then sets {@link #groupKey} to the
  * current key.
  */
 private boolean sameAsPreviousKey(DeferredObject[] currentKey)
     throws HiveException {
   if (null == currentKey && null == groupKey) {
     return true;
   }
   String[] previousKey = groupKey;
   copy(currentKey);
   if (null == groupKey && null != previousKey) {
     return false;
   }
   if (null != groupKey && null == previousKey) {
     return false;
   }
   if (groupKey.length != previousKey.length) {
     return false;
   }
   for (int index = 0; index < previousKey.length; index++) {
     if (!groupKey[index].equals(previousKey[index])) {
       return false;
     }
   }
   return true;
 }

 /**
  * Copies the given key to {@link #groupKey} for future
  * comparisons.
  */
 private void copy(DeferredObject[] currentKey)
     throws HiveException {
   if (null == currentKey) {
     groupKey = null;
   } else {
     groupKey = new String[currentKey.length];
     for (int index = 0; index < currentKey.length; index++) {
       groupKey[index] = String.valueOf(currentKey[index].get());
     }
   }
 }

 @Override
 public String getDisplayString(String[] children) {
   StringBuilder sb = new StringBuilder();
   sb.append("PsuedoRank (");
   for (int i = 0; i < children.length; i++) {
     if (i > 0) {
       sb.append(", ");
     }
     sb.append(children[i]);
   }
   sb.append(")");
   return sb.toString();
 }
}

Although Hadoop clusters are designed to scale from 10 to 10,000 nodes, sometimes deployment-specific requirements involve running more than one filesystem or JobTracker. At M6D, we have such requirements, for example we have hourly and daily process reports using Hadoop and Hive that are business critical and must complete in a timely manner. However our systems also support data science and sales engineers that periodically run ad hoc reporting. While using the fair share scheduler and capacity scheduler meets many of our requirements, we need more isolation than schedulers can provide. Also, because HDFS has no snapshot or incremental backup type features, we require a solution that will prevent an accidental delete or drop table operations from destroying data.

Our solution is to run two distinct Hadoop deployments. Data can have a replication factor of two or three on the primary deployment and additionally be replicated to a second deployment. This decision allows us to have guaranteed resources dedicated to our time-sensitive production process as well as our ad hoc users. Additionally, we protected against any accidental drop tables or data deletes. This design does incur some overhead in having to administer two deployments and setup and administer the replication processes, but this overhead is justified in our case.

Our two deployments are known as production and research. They each have their own dedicated Data Nodes and Task Trackers. Each NameNode and JobTracker is a failover setup using DRBD and Linux-HA. Both deployments are on the same switching network (Tables 23-1 and 23-2).

hive> set fs.default.name;
fs.default.name=hdfs://hdfs.hadoop.pvt:54310
hive> set mapred.job.tracker;
mapred.job.tracker=jt.hadoop.pvt:54311
hive> describe extended zz_mid_set;
OK
adv_spend_id    int
transaction_id  bigint
time    string
client_id       bigint
visit_info      string
event_type      tinyint
level   int

location:hdfs://hdfs.hadoop.pvt:54310/user/hive/warehouse/zz_mid_set
Time taken: 0.063 seconds
hive> select count(1) from zz_mid_set;
1795928

hive> set fs.default.name;
fs.default.name=hdfs://rs01.hadoop.pvt:34310
hive> set mapred.job.tracker;
mapred.job.tracker=rjt.hadoop.pvt:34311
hive> CREATE TABLE EXTERNAL table_in_another_cluster
( adv_spend_id int, transaction_id bigint, time string, client_id bigint,
visit_info string, event_type tinyint, level int)
LOCATION 'hdfs://hdfs.hadoop.pvt:54310/user/hive/warehouse/zz_mid_set';
hive> select count(*) FROM table_in_another_cluster;
1795928

/replication/default.fracture_act/hit_date=20110304,mid=3000

 public void run(){
    while (goOn){
      Path base = new Path(pathToConsume);
      FileStatus [] children = srcFs.listStatus(base);
      for (FileStatus child: children){
        try {
          openHiveService();
          String db = child.getPath().getName().split("\.")[0];
          String hiveTable = child.getPath().getName().split("\.")[1];
          Table table = srcHive.client.get_table(db, hiveTable);
          if (table == null){
            throw new RuntimeException(db+" "+hiveTable+
              " not found in source metastore");
          }
          Table tableR = destHive.client.get_table(db,hiveTable);
          if (tableR == null){
            throw new RuntimeException(db+" "+hiveTable+
              " not found in dest metastore");
          }

          URI localTable = new URI(tableR.getSd().getLocation());
          FileStatus [] partitions = srcFs.listStatus(child.getPath());
          for (FileStatus partition : partitions){
            try {
              String replaced = partition.getPath().getName()
              .replace(",", "/").replace("'","");
              Partition p = srcHive.client.get_partition_by_name(
              db, hiveTable, replaced);
        URI partUri = new URI(p.getSd().getLocation());
        String path = partUri.getPath();
              DistCp distCp = new DistCp(destConf.conf);
              String thdfile = "/tmp/replicator_distcp";
              Path tmpPath = new Path(thdfile);
        destFs.delete(tmpPath,true);
              if (destFs.exists( new Path(localTable.getScheme()+
                "://"+localTable.getHost()+":"+localTable.getPort()+path) ) ){
                throw new RuntimeException("Target path already exists "
                  +localTable.getScheme()+"://"+localTable.getHost()+
                  ":"+localTable.getPort()+path );
              }

              String [] dargs = new String [4];
              dargs[0]="-log";
              dargs[1]=localTable.getScheme()+"://"+localTable.getHost()+":"+
                localTable.getPort()+thdfile;
              dargs[2]=p.getSd().getLocation();
              dargs[3]=localTable.getScheme()+"://"+localTable.getHost()+":"+
                localTable.getPort()+path;
              int result =ToolRunner.run(distCp,dargs);
              if (result != 0){
                throw new RuntimeException("DistCP failed "+ dargs[2] +" "+dargs[3]);
              }

              String HQL = "ALTER TABLE "+hiveTable+
                  " ADD PARTITION ("+partition.getPath().getName()
                +") LOCATION '"+path+"'";
              destHive.client.execute("SET hive.support.concurrency=false");
              destHive.client.execute("USE "+db);
              destHive.client.execute(HQL);
              String [] results=destHive.client.fetchAll();
              srcFs.delete(partition.getPath(),true);
            } catch (Exception ex){
              ex.printStackTrace();
            }
          } // for each partition
        } catch (Exception ex) {
          //error(ex);
          ex.printStackTrace();
        }
      } // for each table
      closeHiveService();
      Thread.sleep(60L*1000L);
    } // end run loop
  } // end run

Sometimes, when you’re trying to aggregate your traffic, it can be tricky to tell where it’s actually coming from, especially for traffic coming from elsewhere in your site. If you have a site with a lot of URLs with different structures, you can’t simply check that the referrer URLs match the landing page.

Host = PARSE_URL(my_url, ‘HOST’’)

ri_page_url STRING
ri_referrer_url STRING

SELECT ri_page_url, ri_referrer_url,
  CASE
    WHEN ri_referrer_url is NULL or ri_referrer_url = ‘’ THEN ‘DIRECT’
    WHEN ri_referrer_url is in (‘mysite1.com’,’mysite2.com’,’mysite3.com’) THEN ‘INSITE’
    ELSE ri_referrer_url
  END as ri_referrer_url_classed
FROM
  referrer_identification;

su_site_id INT
su_url STRING

ri_site_id INT

SELECT
  c.c_page_url as ri_page_url,
  c.c_site_id as ri_site_id,
  CASE
    WHEN c.c_referrer_url is NULL or c.c_referrer_url = ‘’ THEN ‘DIRECT’
    WHEN c.c_insite_referrer_flags > 0 THEN ‘INSITE’
    ELSE c.c_referrer_url
  END as ri_referrer_url_classed
FROM
(SELECT
  a.a_page_url as c_page_url,
  a.a_referrer_url as c_referrer_url,
  a.a_site_id as c_site_id,
  SUM(IF(b.b_url <> ‘’, 1, 0)) as c_insite_referrer_flags
FROM
(SELECT
  ri_page_url as a_page_url,
  ri_referrer_url as a_referrer_url,
  ri_site_id as a_site_id
FROM
  referrer_identification
) a
LEFT OUTER JOIN
(SELECT
  su_site_id as b_site_id,
  su_url as b_url
FROM
  site_url
) b
ON
  a.a_site_id = b.b_site_id and
  a.a_referrer_url = b.b_url
) c

Let’s say you want to calculate the number of unique visitors you have to your site/network/whatever. We’ll use a ridiculously simple schema for our hypothetical table, daily_users:

du_user_id STRING
du_date STRING

However, if you have too many users and not enough machines in your cluster, it might begin to have trouble counting users over a month:

SELECT
 COUNT(DISTINCT du_user_id)
FROM
 daily_users
WHERE
  du_date >= ‘2012-03-01’ and
  du_date <= ‘2012-03-31’

In all likelihood, your cluster is probably able to make it through the map phase without too much problems, but starts having issues around the reduce phase. The problem is that it’s able to access all the records but it can’t count them all at once. Of course, you can’t count them day by day, either, because there might be some redundancies.

CREATE TABLE daily_users_deduped (dud_user_id STRING)
PARTITIONED BY (dud_date STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘t’;

INSERT OVERWRITE TABLE daily_users_deduped
PARTITION (dud_date = ‘:date:’)

SELECT DISTINCT
  du_user_id
FROM
  daily_users
WHERE
  du_date = ‘:date:’

start_date = ‘2012-03-01’
end_date = ‘2012-03-31’

for date in date_range(start_date, end_date):
  femp_file = modify_temp_file(‘mj_01.sql’,{‘:date:’:my_date})
  fire_query(temp_file)
  delete(temp_file)

SELECT
  COUNT(DISTINCT (dud_uuid)
FROM
  daily_users_deduped

For analyzing web traffic, we often want to be able to measure engagement based on various criteria. One way is to break up user behavior into sessions, chunks of activity that represent a single “use.” A user might come to your site several times a day, a few days a month, but each visit is certainly not the same.

So, what is a session? One definition is a string of activity, not separated by more than 30 minutes. That is, if you go to your first page, wait five minutes, go to the second page, it’s the same session. Wait 30 minutes exactly until the third page, still the same session. Wait 31 minutes until that fourth page, and the session will be broken; rather than the fourth pageview, it would be the first page of the second session.

Once we’ve got these broken out, we can look at properties of the session to see what happened. The ubiquitous case is to compare referrers to your page by session length. So, we might want to find out if Google or Facebook give better engagement on your site, which we might measure by session length.

At first glance, this seems perfect for an iterative process. For each pageview, keep counting backwards until you find the page that was first. But Hive isn’t iterative.

You can, however, figure it out. I like to break this into four phases.

This leaves a table where each row represents a full session, which you can then query for whatever you want to find out.

st_user_id STRING
st_pageview_id STRING
st_page_url STRING
st_referrer_url STRING
st_timestamp DOUBLE

CREATE TABLE sessionization_step_one_origins AS

SELECT
  c.c_user_id as ssoo_user_id,
  c.c_pageview_id as ssoo_pageview_id,
  c.c_timestamp as ssoo_timestamp
FROM
        (SELECT
    a.a_user_id as c_user_id,
    a.a_pageview_id as c_pageview_id,
    a.a_timestamp as c.c_timestamp,
    SUM(IF(a.a_timestamp + 1800 >= b.b_timestamp AND
        a.a_timestamp < b.b_timestamp,1,0)) AS c_nonorigin_flags
  FROM
    (SELECT
      st_user_id as a_user_id,
      st_pageview_id as a_pageview_id,
      st_timestamp as a_timestamp
    FROM
      session_test
    ) a
    JOIN
    (SELECT
      st_user_id as b_user_id,
      st_timestamp as b_timestamp
    FROM
      session_test
    ) b
    ON
      a.a_user_id = b.b_user_id
  GROUP BY
    a.a_user_id,
    a.a_pageview_id,
    a.a_timestamp
  ) c
WHERE
  c.c_nonorigin_flags

SUM(IF(a.a_timestamp + 1800 >= b.b_timestamp AND
    a.a_timestamp < b.b_timestamp,1,0)) as c_nonorigin_flags

ssoo_user_id STRING
ssoo_pageview_id STRING
ssoo_timestamp DOUBLE

CREATE TABLE sessionization_step_two_origin_identification AS

SELECT
  c.c_user_id as sstoi_user_id,
  c.c_pageview_id as sstoi_pageview_id,
  d.d_pageview_id as sstoi_origin_pageview_id
FROM
(SELECT
  a.a_user_id as c_user_id,
  a.a_pageview_id as c_pageview_id,
  MAX(IF(a.a_timestamp >= b.b_timestamp, b.b_timestamp, NULL)) as c_origin_timestamp
FROM
(SELECT
  st_user_id as a_user_id,
  st_pageview_id as a_pageview_id,
  st_timestamp as a_timestamp
FROM
  session_test
) a
JOIN
(SELECT
  ssoo_user_id as b_user_id,
  ssoo_timestamp as b_timestamp
FROM
  sessionization_step_one_origins
) b
ON
  a.a_user_id = b.b_user_id
GROUP BY
  a.a_user_id,
  a.a_pageview_id
) c
JOIN
(SELECT
  ssoo_usr_id as d_user_id,
  ssoo_pageview_id as d_pageview_id,
  ssoo_timestamp as d_timestamp
FROM
  sessionization_step_one_origins
) d
ON
  c.c_user_id = d.d_user_id and
  c.c_origin_timestamp = d.d_timestamp

MAX(IF(a.a_timestamp >= b.b_timestamp, b.b_timestamp, NULL)) as c_origin_timestamp

sstoi_user_id STRING
sstoi_pageview_id STRING
sstoi_origin_pageview_id STRING

CREATE TABLE sessionization_step_three_origin_aggregation AS

SELECT
  a.a_user_id as sstoa_user_id,
  a.a_origin_pageview_id as sstoa_origin_pageview_id,
  COUNT(1) as sstoa_pageview_count
FROM
  (SELECT
    ssoo_user_id  as a_user_id
    ssoo_pageview_id as a_origin_pageview_id
  FROM
    sessionization_step_one_origins
  ) a
  JOIN
  (SELECT
    sstoi_user_id as b_user_id,
    sstoi_origin_pageview_id as b_origin_pageview_id
  FROM
    sessionization_step_two_origin_identification
  ) b
ON
  a.a_user_id = b.b_user_id and
  a.a_origin_pageview_id = b.b_origin_pageview_id
GROUP BY
  a.a_user_id,
  a.a_origin_pageview_id

CREATE TABLE sessionization_step_four_qualitative_labeling

SELECT
  a.a_user_id as ssfql_user_id,
  a.a_origin_pageview_id as ssfql_origin_pageview_id,
  b.b_timestamp as ssfql_timestamp,
  b.b_page_url as ssfql_page_url,
  b.b_referrer_url as ssfql_referrer_url,
  a.a_pageview_count as ssqfl_pageview_count
(SELECT
  sstoa_user_id as a_user_id,
  sstoa_origin_pageview_id as a_origin_pageview_id,
  sstoa_pageview_count as a_pageview_count
FROM
  sessionization_step_three_origin_aggregation
) a
JOIN
(SELECT
  st_user_id as b_user_id,
  st_pageview_id as b_pageview_id,
  st_page_url as b_page_url,
  st_referrer_url as b_referrer_url,
  st_timestamp as b_timestamp
FROM
  session_test
) b
ON
  a.a_user_id = b.b_user_id and
  a.a_origin_pageview_id = b.b_pageview_id

SELECT
  PARSE_URL(ssfql_referrer_url, ‘HOST’) as referrer_host,
  COUNT(1) as session_count,
  AVG(ssfql_pageview_count) as avg_pvs_per_session,
  SUM(ssfq_pageview_count)/COUNT(1) as weighted_avg_pvs_per_session,
  MAX(ssfql_pageview_count) as max_pvs_per_session,
  MIN(ssfql_pageview_count) as min_pvs_per_session,
  COUNT(DISTINCT ssfql_usr_id) as unique_users
FROM
  sessionization_step_three_origin_aggregation
GROUP BY
  PARSE_URL(ssfql_referrer_url, ‘HOST’) as referrer_host

by Chris A. Mattmann, Paul Zimdars, Cameron Goodale, Andrew F. Hart, Jinwon Kim, Duane Waliser, Peter Lean

Since 2009, our team at NASA’s Jet Propulsion Laboratory (JPL) has actively led the development of a Regional Climate Model Evaluation System (RCMES). The system, originally funded under the American Recovery and Reinvestment Act (ARRA) has the following goals:

Figure 23-2 shows the architecture and data flow of the Regional Climate Model Evaluation System

Figure 23-2. JPL Architecture Diagram.

In support of these goals, we have constructed a multifaceted system shown in Figure 23-2. Reading the diagram from left to right, available reference data sets from observations and assimilations, especially from satellite-based remote sensing, enters the system according to the desired climate parameters useful for climate model evaluation. Those parameters are stored in various mission data sets, and those data sets are housed in several external repositories, eventually fed into the database component (RCMED: Regional Climate Model Evaluation Database) of RCMES.

As an example, AIRS is NASA’s Atmospheric Infrared Sounder and provides parameters including surface air temperature, temperature, and geopotential; MODIS is NASA’s Moderate Imaging Spectroradiometer and provides parameters including cloud fraction; and TRMM is NASA’s Tropical Rainfall Measurement Mission and provides parameters including monthly precipitation. This information is summarized in our RCMES system website parameter table, http://rcmes.jpl.nasa.gov/rcmed/parameters/, and shown in Figure 23-3.

Figure 23-3. JPL Physical Architecture Diagram

Data sets are loaded into the RCMED using the Apache OODT extractor framework and the desired parameters, their values, spatial and temporal constraints (and optionally height) are loaded and potentially transformed (e.g., normalized, put on same coordinate system, converted from unit values) into a MySQL database. The data loaded into that MySQL database, RCMED, is exposed to external clients via a Space/Time query and subsetting web service; the description of which is a topic of a separate study. For all intents and purposes, it provides the same capabilities that the OPeNDAP technology does.

The right side of the diagram shows the Regional Climate Model Evaluation Toolkit (RCMET). It provides users with the ability to take in the reference data from RCMED and climate model output data produced elsewhere and to re-grid these datasets in order to match them spatially and temporally in preparation for the comparison of the reference and model data for the evaluation of model output against the user-selected reference data. At that point, the system allows for seasonal cycle compositing (e.g., all Januaries, or all summer months for N years), and for preparing the data for eventual metrics calculation, that is, comparison of the values of the model output against the remote sensing data observation parameters and their values. The system supports several metrics, such as bias computation, Root Mean Squared Error (RMSE), and the generation of relevant visualizations, including traditional plots and Taylor diagrams for science use/decision making.

So, where does Hive come in to play? After loading 6 billion rows of (latitude, longitude, time, data point value, height) tuples into MySQL, the system fell down and experienced data loss. This is probably due in part to our naïve strategy of storing all of the data points in a single table. Over time, we evolved this strategy to break tables down by dataset and by parameter, which helped but added needless overhead that we didn’t want to spend cycles engineering around.

Instead, we decided to experiment with the Apache Hive technology. We installed Hive 0.5+20 using CDHv3 and Apache Hadoop (0.20.2+320). CDHv3 came with a number of other relevant tools including Sqoop, and Hue, which we leveraged in our architecture, shown in the bottom portion of Figure 23-3.

We used Apache Sqoop to dump out the data into Hive, and then wrote an Apache OODT wrapper that queried Hive for the data by Space/Time and provided it back to the RCMET and other users (shown in the middle portion of Figure 23-2). The full architecture for the RCMES cluster is shown in Figure 23-3. We had five machines, including a master/slave configuration as shown in the diagram, connected by a private network running GigE.

During the migration of data from MySQL to Hive, we experienced slow response times while doing simple tasks such as a count DB query (e.g., hive> select count(datapoint_id) from dataPoint;). We initially loaded up around 2.5 billion data points in a single table and noticed that on our machine configuration, Hive took approximately 5–6 minutes to do a count of these 2.5 billion records (15–17 minutes for the full 6.8 billion records). The reduce portion was fast (we were experiencing a single reduce phase since we were using a count * benchmark) but the map stage took the remainder of the time (~95%). Our system at the time consisted of six (4 x quad-core) systems with approximately 24 GB of RAM each (all of the machines shown in Figure 23-3, plus another “borrowed machine” of similar class from another cluster).

We attempted to add more nodes, increase map tasktrackers (many different #s), change DFS block size (32 M, 64 M, 128 MB, 256 M), leverage LZO compression, and alter many other configuration variables (io.sort.factor, io.sort.mb) without much success in lowering the time to complete the count. We did notice a high I/O wait on the nodes no matter how many task trackers we ran. The size of the database was approximately ~200GB and with MySQL it took a few seconds to do both the 2.5 billion and 6.7 billion count.

Members of the Hive community jumped in and provided us with insight, ranging from mentioning that HDFS read speed is about 60 MB/sec comparing to about 1 GB/sec on local disk, depending of course on network speed, and namenode workload. The numbers suggested by the community member suggested that we needed roughly 16 mappers in the Hadoop job to match with the I/O performance of a local non-Hadoop task. In addition, Hive community members suggested that we increase the number of mappers (increase parallelism) by reducing the split size (input size) for each mapper, noting we should examine the following parameters: mapred.min.split.size, mapred.max.split.size, mapred.min.split.size.per.rack, and mapred.min.split.size.per.node, and suggesting that the parameters should be set to a value of 64 MB. Finally, the community suggested that we look at a benchmark that only counts rows by using count(1) instead of count (datapoint_id), as the latter is faster since no column reference means no decompression and deserialization, e.g., if you store your table in RCFile format.

Based on the above feedback, we were able to tune our Hive cluster for RCMES to respond to a count query benchmark, and to a space/time query from RCMET for billions of rows in under 15 seconds, using the above-mentioned resources, making Hive a viable and great choice for our system architecture.

Prior to 2008, Photobucket didn’t have a dedicated analytics system in-house. Questions from the business users were run across hundreds of MySQL instances and the results aggregated manually in Excel.

In 2008, Photobucket embarked on implementing its first data warehouse dedicated to answering the increasingly complex data questions being asked by a fast-growing company.

The first iteration of the data warehouse was built using an open source system with a Java SQL optimizer and a set of underlying PostGreSQL databases. The previous system worked well into 2009, but the shortcomings in the architecture became quickly evident. Working data sets quickly became larger than the available memory; coupled with the difficulty in repartitioning the data across the PostGreSQL nodes forced us to scale up when we really wanted to scale out.

In 2009, we started to investigate systems that would allow us to scale out, as the amount of data continued to grow and still meet our SLA with the business users. Hadoop quickly became the favorite for consuming and analyzing the terabytes of data generated daily by the system, but the difficulty of writing MapReduce programs for simple ad hoc questions became a negative factor for full implementation. Thankfully, Facebook open sourced Hive a few weeks later and the barriers to efficiently answering ad hoc business questions were quickly smashed.

Hive demonstrates many advantages over the previous warehouse implementation. Here are a few examples of why we chose Hadoop and Hive:

Dell R410, 4 × 2 TB drives with 24 GB RAM for the data nodes, and Dell R610, 2 × 146 GB (RAID 10) drives with 24 GB RAM for the management hardware.

The primary goal of Hive at Photobucket is to provide answers about business functions, system performance, and user activity. To meet these needs, we store nightly dumps of MySQL data sets from across hundreds of servers, terabytes of logfiles from web servers and custom log formats ingested through Flume. This data helps support many groups throughout the company, such as executive management, advertising, customer support, product development, and operations just to name a few. For historical data, we keep the partition of all data created on the first day of the month for MySQL data and 30+ days of log files. Photobucket uses a custom ETL framework for migrating MySQL data into Hive. Log file data is streamed into HDFS using Flume and picked up by scheduled Hive processes.

Executive management relies on Hadoop to provide reports surrounding the general health of the business. Hive allows us to parse structured database data and unstructured click stream data and distill the data into a format requested by the business stakeholder.

Advertising operations uses Hive to sift through historical data for forecast and define quotas for ad targeting.

Product development is far and away the group generating the largest number of ad hoc queries. As with any user base, segments change and evolve over time. Hive is important because it allows us to run A/B tests across current and historical data to gauge relevancy of new products in a quickly changing user environment.

Providing our users with a best-in-class system is the most important goal at Photobucket. From an operations perspective, Hive is used to generate rollup data partitioned across multiple dimensions. Knowing the most popular media, users, and referring domains is important for many levels across the company. Controlling expenses is important to any organization. A single user can quickly consume large amounts of system resources, significantly increasing monthly expenditures. Hive is used to identify and analyze rogue users; to determine which ones are within our Terms of Service and which are not. Operations also uses Hive to run A/B tests defining new hardware requirements and generating ROI calculations. Hive’s ability to abstract users from underlying MapReduce code means questions can be answered in hours or days instead of weeks.

By Nanda Vijaydev

For over 18 months, Karmasphere has been engaged with a fast-growing number of companies who adopted Hadoop and immediately gravitated towards Hive as the optimal way for teams of analysts and business users to use existing SQL skills with the Hadoop environment. The first part of this chapter provides use case techniques that we’ve seen used repeatedly in customer environments to advance Hive-based analytics.

The use case examples we cover are:

As companies we’ve worked with plan for and move into production use of Hive, they look for incremental capabilities that make Hive-based access to Hadoop even easier to use, more productive, more powerful, and available to more people in their organization. When they wire Hadoop and Hive into their existing data architectures, they also want to enable results from Hive queries to be systematized, shared and integrated with other data stores, spreadsheets, BI tools, and reporting systems.

In particular, companies have asked for: