Runing Spark Unit Test on Windows 7

It's common situation in enterprises when developers are working on Windows platform. When you are working with Hadoop, it sounds as a f**ing shit, but this is a fact.

Recently, I switched in a favor of Spark instead of traditional MapReduce paradigm and was need to implement some kind of unit/integration testing... of course, it was need to work under Windows 7.

I've written very simple test: run ETL in-memory, without touching Hadoop at all (in future, I'd like to read input from local filesystem):

@Test
def testETL() = {
    val conf = new SparkConf()
    val sc = new SparkContext("local", "test", conf)
    try {
        val etl = new IxtoolsDailyAgg() // empty constructor

        val data = sc.parallelize(List("in1", "in2", "in3"))

        etl.etl(data) // rdd transformation, no access to SparkContext or Hadoop
        Assert.assertTrue(true)
    } finally {
        if(sc != null)
            sc.stop()
    }
}

Bum! I got exception:

java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.
 at org.apache.hadoop.util.Shell.getQualifiedBinPath(Shell.java:318)
 at org.apache.hadoop.util.Shell.getWinUtilsPath(Shell.java:333)
 at org.apache.hadoop.util.Shell.<clinit>(Shell.java:326)
 at org.apache.hadoop.util.StringUtils.<clinit>(StringUtils.java:76)
 at org.apache.hadoop.security.Groups.parseStaticMapping(Groups.java:93)
 at org.apache.hadoop.security.Groups.<init>(Groups.java:77)
 at org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:240)
 at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:255)
 at org.apache.hadoop.security.UserGroupInformation.setConfiguration(UserGroupInformation.java:283)
 at org.apache.spark.deploy.SparkHadoopUtil.<init>(SparkHadoopUtil.scala:36)
 at org.apache.spark.deploy.SparkHadoopUtil$.<init>(SparkHadoopUtil.scala:109)
 at org.apache.spark.deploy.SparkHadoopUtil$.<clinit>(SparkHadoopUtil.scala)
 at org.apache.spark.SparkContext.<init>(SparkContext.scala:228)
 at org.apache.spark.SparkContext.<init>(SparkContext.scala:97)

What?

org.apache.hadoop.util.Shell.(Shell.java:326)

I swear, I didn't use Hadoop in my code!
Unfortunately, Hadoop configuration is initialized together with SparkContext :( no way to omit it...
I was recommended to install HDP on Windows, but I hate this idea...

I tried the most stupid idea - provide winutils.exe... I hope, it's only the check of environment and Hadoop functionality won't be used if I don't touch it.
So, I downloaded winutils.exe from msdn (msdn still helpful even for hadooper), put it to created directory d:\winutil\bin and then add

System.setProperty("hadoop.home.dir", "d:\\winutil\\") 

at the beginning of my unit test

Hue Notifier for Hadoop goes wild

Several months ago I developed Chrome browser plugin for my own needs. As a Hadoop engineer I faced with one problem everyday. I run a lot of Hive/Pig jobs simultaneously and they take a lot of time (from several minutes to several hours). So, I had mission to check job completion by walking Hue's pages in my browser. Well, it was 1) irritate, 2) draw away from coding...

As solution I developed Hue Notifier for Hadoop plugin for Google Chrome. It "monitors" state of job and inform you about completion similar to GMail informs about new mail (pop-up over all windows). I have a quite limited knowledge of JavaScript and it has been first time I wrote browser plugin... so, I'm absolutely sure it might be improved. I tested it with Hue delivered with Cloudera 4.3 and Cloudera 5 as well as HDP2.0. The most irritating issue w/ my code: Chrome Notification must be enabled manually before start using plugin :(

The source code is generally available at GitHub under this repository. You are welcome to fork and improve this one. Or, if you wish just to contribute, ping me and I will grant access (and push changes to Google Play afterwards).

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