Building BuilData ETL with Hive and Oozie

Perhaps, Hive is the most successful component of today's Hadoop infrastructure. It provides simple and efficient way of creating Hadoop-based data processing jobs with comfortable SQL-like language. But, in contract to Pig, it's not a workflow-friendly language and requires additional effort to create a real multi-step ETL.
Oozie was created to eliminate workflow/scheduling issues and, obvious, may be used to create ETL and naturally engages Hive.

Spark on HDP2

There is my first experience with Apache Spark, running it on Hadoop. I faced in several issues during running my piece of code.
To be honest, I started with Cloudera CDH5 distribution, they promised Spark was already added and usage will be simple. But no luck in fact, it doesn't work at all - even on local machine with their spark-cloudera jar. I didn't want to waste my time, so I just downloaded spark distro to HDP2.
First of all, let start Spark in standalone mode, according to documentation:

# start master
./sbin/start-master.sh

# pick up in the log output spark://IP:PORT
# and than run worker on each node
./bin/spark-class org.apache.spark.deploy.worker.Worker spark://IP:PORT

# more documentation available here https://spark.apache.org/docs/0.9.0/spark-standalone.html

After that I wrote some amount of Scala code, in fact to just count hardcoded words in document:

package experiment

import org.apache.spark.{SparkConf, SparkContext}

object SimpleApp {

  def main(args: Array[String]) {
    val logFile = args(0)  
  val conf = new SparkConf()
      .setMaster("local")
      .setAppName("My Spark application")
      .set("spark.executor.memory", "1g")
  val sc = new SparkContext(conf)


  // hdfs:///user/hue/input.txt
    val logData = sc.textFile(logFile, 2).cache()
    val numAs = logData.filter(line => line.contains("London")).count()
    val numBs = logData.filter(line =>; line.contains("Lviv")).count()
    println("Lines with London: %s, Lines with Lviv: %s".format(numAs, numBs))
}

XQuery on Hadoop

Java is mother language for the most of Hadoop engineers. In recent years, Python became popular, R is used by data scientist on Hadoop. Pig Latin and HiveQL is de-facto the mainstream languages for Hadoop now days. Oracle decided to not stop on that and gives possibility to write MapReduce jobs in XQuery! Unbelievable, xml-fans must be happy :)

Let's review simple example.

First of all, Oracle BigData Lite VM must be downloaded (for free, but it takes 25Gb on disk).

After installation, test dataset must be create. I put 2 files to directory on HDFS /user/oracle/xquery/input with sample dataset about access to website. The example of content is:
2013-10-28T06:00:00, chrome, index.html, 200
2013-10-28T08:30:02, firefox, index.html, 200
2013-10-28T08:32:50, ie9, about.html, 200

Next step: create XQuery script (my_xquery.xq) to process data (simple grouping by date of visiting page)

import module "oxh:text";

for $line in text:collection("/user/oracle/xquery/input/*.txt")
let $split := fn:tokenize($line, "\s*,\s*")
let $time := xs:dateTime($split[1])
let $day := xs:date($time)
group by $day
return text:put($day || ", " || fn:count($line))

Now script is ready to be run, execute from command line:

hadoop jar $OXH_HOME/lib/oxh.jar my_xquery.xq -output /user/oracle/xquery/output -clean -ls

Options:
-output specify output directory
-clean remove output directory if exists
-ls list the content of output directory after run

Here is the result:

That's it, XQuery was translated to MapReduce (similar to Pig Latin or HiveQL). This functionality is the part of Oracle BigData Connectors for Hadoop and more information with examples might be read here

How to write good unit test for Hadoop MapReduce?

Without a doubt, there is avery common situation when UnitTest (or IntegrationTest) is required to test functionality of MapReduce job. This approach perfect fit TDD, moreover, it gives opportunity to develop MapReduce jobs faster, because there is no needs to redeploy jar on a cluster each time and debugging is easy to use.

The first line of defence is MRUnit. Great framework for unit testing, input/output format independent with possibility to run/test map and reduce functions separately. Unfortunately, this framework has a several meaningful drawbacks. For example, no access to MR counters, or during the MR test only one Mapper allowed.

Local execution mode may be used to overcome MRUnit limitations or create integration test for mapreduce job. Let's assume there is runnable MapReduce tool with several input sources (mappers) and reducer:

public class ExampleMrDriver extends Configured implements Tool {

 public  Job createMRJob(Configuration conf) throws IOException {...}

 @Override
    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
        Job job = createMRJob(conf);
        return job.waitForCompletion(true) ? 0 : -1;
    }


 public static void main(String[] args) {
        try {
         // run job in a Oozie-friendly manner
            int status = ToolRunner.run(new ExampleMrDriver(), args);
            if(status!=0) {
                System.exit(status);
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

}

Nice integration test (or unit, call and use it as you like) for this Hadoop MapReduce a listed bellow:

private String outputDir;

@BeforeClass
public void createTmpDir() throws IOException {
    outputDir = System.getProperty("java.io.tmpdir"); + "output";
}

@Test
public void test() throws Exception {
    JobConf jobConf = new JobConf();
    jobConf.set("fs.default.name", "file:///"); 
    jobConf.set("mapred.job.tracker", "local"); // local mode
    jobConf.set("mapred.reduce.task", "1"); // only one file is required in output

    // create file w/ input content per mapper in test/resource folder
    jobConf.set("input.dir.2", this.getClass().getResource("/mr/inpu1").getPath());
    jobConf.set("input.dir.1", this.getClass().getResource("/mr/input2").getPath());
    jobConf.set("input.dir.3", this.getClass().getResource("/mr/input3").getPath());
    // expected output will be placed here
    jobConf.set("output.dir", outputDir);

    ExampleMrDriver driver = new ExampleMrDriver();
    driver.setConf(jobConf);
    int exitCode = driver.run(new String[]{});

    Assert.assertEquals(0, exitCode);

    // check content of output file, counters, etc
}

@AfterClass
public void tearDown() throws IOException {
    new File(outputDir).delete();
}

Predicted Age of Abalone based on physical measurements

Abalone dataset is freely available at UCI Machine Learning Repository since 1995. It contains result of abalone research in Australia. Predicting the age of abalone from physical measurements. The age of abalone is determined by cutting the shell through the cone, staining it, and counting the number of rings through a microscope - a boring and time-consuming task. Other measurements, which are easier to obtain, are used to predict the age. Definitely, the task is more complex in the real conditions and further information, such as weather patterns and location (hence food availability) may be required to solve the problem.

So, Age ~ Rings and must be predicted from the set of different measures as Diameter, Weight, Height, Length, etc. It is supervised learning task, because of the dataset with relation Result~Features is provided. Simple check shows numbers of rings from 1 to 29 and it is huge range for classification. Another supervised learning algorithm is a linear regression. 

EDA (exploratory data analysis) is a first step before building any model and there is the code for loading dataset into memory and plotting several relations, for example Rings~Diameter

library(ggplot2)
 
# read dataset from local file
abalone <- read.csv("/Users/kostya/Downloads/abalone.data.csv", header=F)
 
# set names for dataframe columns
colnames(abalone) <- c('Sex', 'Length', 'Diameter', 'Height', 'WholeWeight', 'ShuckedWeight',
                    'VisceraWeight', 'ShellWeight', 'Rings')
 
# plot histogram
hist(abalone$Rings, freq=F)
 
# depicture all charts on one plot
qplot(Diameter, Rings, data=abalone, geom=c("point", "smooth"), method="lm", color=Sex, se=F)

This image (as well as other relations like Rings~WholeWeight, etc) shows pretty well difference relations for each sex and the first thought is to apply different regression for each 'sex' or use 'sex' as a factor.

For example, go on with different regression models, we need to construct formula by investigating each relations. For example, there is Rings~WholeWeight relation 

# plot each sex on different plot
ggplot(abalone, aes(VisceraWeight, Rings)) + 
  geom_jitter(alpha=0.25) + 
  geom_smooth(method=lm, se=FALSE) +
  facet_grid(. ~ Sex)

Obvious, that for Male and Infant relations has logarithmic trend and it will be logically to add 'log' in formula. 

summary(lm(Rings~Length+I(Diameter^2)+log(WholeWeight)+log(ShellWeight)+log(ShuckedWeight)
           +Height+VisceraWeight, data=subset(abalone, Sex %in% 'I'))  )
 
summary(lm(Rings~Length+I(Diameter^2)+log(WholeWeight)+log(ShellWeight)+ShuckedWeight
           +Height+VisceraWeight, data=subset(abalone, Sex %in% 'M'))  )
 
summary(lm(Rings~Length+I(Diameter^2)+WholeWeight+ShellWeight+ShuckedWeight
           +Height+VisceraWeight, data=subset(abalone, Sex %in% 'F'))  )

As result the next formula may be constructed to predict number of rings for Infant based on coefficient of linear regression:
Rings= 8.5398 - 7.6755*Length + 8.7707*Diameter^2 + 1.4837*log(WholeWeight) + 2.0745*log((ShellWeight) -2.3415*log(ShuckedWeight) + 27.8275*Height + 5.9972*VisceraWeight

As was mentioned in task description Age=Rings+1.5

R connection to Hive

Short instruction how to query Hive from R via JDBC.

First of all install rJava: sudo apt-get install r-cran-rjava
After that install RJDBC package with all dependencies: install.packages("RJDBC",dep=TRUE)
In next step Hadoop libraries for Hive conneections must be added to classpath. The easiest way to do it: copy all jars for pattern /usr/lib/hive/lib/*.jar and /usr/lib/hadoop/*.jar to your classpath on target machine (when RJDBC client is located).

Also, HiveServer must be started, for Cloudera distribution use

hive --service hiveserver2

instead of

sudo service hive-server2 start

(as was mentioned http://www.cloudera.com/content/cloudera-content/cloudera-docs/CDH4/4.2.1/CDH4-Installation-Guide/cdh4ig_topic_18_8.html)
Now it is time to check if HiveServer is running properly, follow the next command line steps:

/usr/lib/hive/bin/beeline
beeline> connect jdbc:hive2://localhost:10000 username password org.apache.hive.jdbc.HiveDriver

Connecting to jdbc:hive2://127.0.0.1:10000/default
Connected to: Hive (version 0.10.0)
Driver: Hive (version 0.10.0-cdh4.3.0)
Transaction isolation: TRANSACTION_REPEATABLE_READ

Finaly, we can write R code to connect Hive and fetch some information

library(RJDBC)
# this is a regular JDBC connection
# jdbc:hive://192.168.0.104:10000/default
drv <- JDBC(driverClass = "org.apache.hive.jdbc.HiveDriver",
            classPath = list.files("/opt/jars/hive",pattern="jar$",full.names=T),
            identifier.quote="`")
conn <- dbConnect(drv, "jdbc:hive2://192.168.0.104:10000/default", "admin", "admin")

r <- dbGetQuery(conn, "select col_1, sum(col_2) from tab2 where id>? group by col_1", "10")

And the result is going to be like:

col_1          _c1
1 false 243808846,65
2  true       486,65

Create Impala DataMart based on Hive backend

Hive queries are slow, hopefully on Cloudera there is possible to create fast accessible Impala DataMart.

Data into Impala table can be populated from Hive table.
The following table is accessible in `default` database:

table tab2 (
  id int,
  col_1 boolean,
  col_2 double)

The following queries can be used to create Impala and Hive tables with the same content (and the difference in the speed of access to these datasets):

Impala	Hive
create table tab5 ( col1 boolean, col2 double) STORED AS PARQUETFILE; insert overwrite tab5 select col_1, sum(col_2) from tab2 group by col_1;	create table tab5h ( col1 boolean, col2 double) STORED AS sequencefile; insert overwrite table tab5h select col_1, sum(col_2) from tab2 group by col_1;