Configure Oozie's Launcher Job

We use Oozie as management application for some of our data processing pipelines. Although the Oozie developers wrote a lot of documentation on Oozie, there are several features and usecases which are covered quite minimalistic by documentation. How to configure the launcher job, for instance, is something I was only able to learn from mailing lists.

The launcher job is used by Oozie to supervise some of its actions, e.g. java or mapreduce actions. The launcher job is executed as a Hadoop job with a single map task and zero reduce tasks. In most cases we do not care much about the launcher. However, there are some situations in which we would like to have some influence on the execution of the launcher job. For example, we wanted to run the complete data processing pipeline with priority VERY_HIGH. Java and mapreduce Oozie actions provide a configuration element which can be populated with arbitrary (with the exception of namenode and jobtracker) Hadoop properties. However, Oozie applies these properties only to the real actions and not to the launcher application. For this purpose, one has to add an oozie.launcher. prefix to the corresponding Hadoop properties.

For the purpose of prioritizing the data processing pipeline with configuration parameters we added the following XML blocks to the configuration elements to all our mapreduce and java actions:

    oozie.launcher.mapred.job.priority
    ${priority}


    mapred.job.priority
    ${priority}

When starting the Oozie workflows, we provide appropriate properties files which contain a priority key with the desired priority setup.

Other useful applications of the oozie.launcher. configuration prefix could be

• to run the launcher job in another queue than the workflow jobs itselves (oozie.launcher.mapred.job.queue.name, see OOZIE-9) or
• to use special java options like increased heap space settings for java actions (oozie.launcher.mapred.child.java.opts)

MultipleOutputFormat and File Handle Limitations

Recently, we used Hadoop for a heavy batch processing job. The job itself was not very special, in fact, the very same job is run on a daily basis to process some sort of data incrementally. The job instances had run fine for several months. Now we wanted to process the data of some months at once and all of a sudden, the processing job died with nasty (and somehow missleading) exceptions. The reduce task logs were filled with lots of stacktraces like

java.io.EOFException
 at java.io.DataInputStream.readByte(DataInputStream.java:250)
 at org.apache.hadoop.io.WritableUtils.readVLong(WritableUtils.java:298)
 at org.apache.hadoop.io.WritableUtils.readVInt(WritableUtils.java:319)
 at org.apache.hadoop.io.Text.readString(Text.java:400)
 at org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.createBlockOutputStream(DFSClient.java:2901)
 at org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.nextBlockOutputStream(DFSClient.java:2826)
 at org.apache.hadoop.hdfs.DFSClient$DFSOutputStream.access$2000(DFSClient.java:2102)
 at org.apache.hadoop.hdfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:2288)

This stacktrace alone did not provide a lot of information about the real exception cause. However, together with the corresponding name node logs, it became clear that the corresponding reduce processes could not open new files for output writing. After realizing that the job is using org.apache.hadoop.mapred.lib.MultipleOutputFormat for output writing, the reason for the failing jobs became clear: file handle limitations. The only question was: which ones? The Linux OS has these limitations and Hadoop's HDFS as well. To make a long story short, we had to increase both of them.

Linux and Open File Limits

Linux limits the number of parallel open files on a per process basis. A given user might only start a certain number of processes (type ulimit -u to see your own limit) in parallel and each of these processes is only allowed to write to a certain number of files in parallel (ulimit -n). The default value for open files is 1024 (at least in debian/ubuntu-flavored distributions). To get around our problem from above, we increased this limit for the user running our Hadoop cluster by editing the /etc/security/limits.conf. To make the machine recognize the new limits, it is necessary to log out and afterwards back in again. However, in our case we do not login as the user hadoop directly, but using the su command. Thus, no new login shell is started and the configuration option would not be recognized. In his blog post, Armin describes how to edit /etc/pamd.d/su in this scenario.

Hadoop and Open File Limits

Hadoop (we use version 0.20.2) has configuration parameters for almost everything. The one to specify the number of parallel files per datanode is named dfs.datanode.max.xcievers. Unfortunately, if not set otherwise, datanodes are at startup equipped with only 256 parallel file handles (see org.apache.hadoop.hdfs.server.datanode.DataXceiverServer class for details). The HBase documentation on the xcievers parameter recommends a value of 4096. As stated before, this parameter is evaluated during datanode startup time. Therefore, it is necessary to configure this parameter in the conf/hdfs-site.xml file on each datanode and restart the cluster afterwards.

Summary

We had to increase both the OS specific limits and the limits in the Hadoop configuration. None of them alone was sufficient. To accomplish the configuration changes, we had to update setting on each datanode machine and to restart the cluster. Afterwards, these exceptions from above were only to be seen on cluster nodes which were not properly updated.

Shutdown MiniDFSCluster and MiniMRCluster Takes Forever

Writing unit tests for Hadoop applications does not need to be more complicated than writing tests for any other Java application.
Usually, I use the following procedure for testing my Hadoop code:

1. Testing the classes as real units.
Each Mapper and each Reducer deserves to be tested in isolation. Spock with its awesome support for mocking and stubbing is a great tool for testing units in isolation. With few 3rd party dependencies it is even possible to mock final classes, bypass the existing constructors of these classes, and inject them into private fields of the units under test => isolation at its best (more on this in another post).
Furthermore, there might be other units like custom Writables, InputFormat and OutputFormat implementations and classes containing the business logic to convert data, and so on. I try to write a Specification (the Spock counterpart of a TestCase) for each non-trivial class.
This should reveal most of the nasty little bugs contained in the business logic of the application. Small side-note: I know, there is a mr-unit test framework provided by Cloudera. Personally, I found Spock to be more powerful and flexible, but as always, this is merely a matter of taste.
2. Next step is to execute complete job roundtrips using the local Hadoop mode, again with Spock. If the application features a command line interface, I set up a simple Specification which executes the main() method of the main driver class and compares some expected output files against the actual output files.
Being able to execute jobs in local mode is great, because it is a fast way to run blackbox test against the map reduce framework. Howerver, since the local mode is somewhat limited, it might be necessary to use a real cluster. For example, in local mode it is not possible to run more than one reducer. This limits the testing capabilities of custom partitioning and grouping code.
3. To execute m/r code on a real cluster, I often use the hadoop-test project (include it with testCompile 'org.apache.hadoop:hadoop-test:0.20.2' in the build.gradle file). This project features implementations of both a distributed filesystem (org.apache.hadoop.hdfs.MiniDFSCluster) and a m/r cluster (org.apache.hadoop.mapred.MiniMRCluster). The most annoying thing about these clusters is the very long startup time, but hey, they're distributed. The remainder of this post is about usage of these clusters.

In case there are several different Hadoop jobs to test or only a single job with different configurations, because of the long startup time, one should think about putting the cluster management code into the static setup methods of the test framework.

static MiniMRCluster mrCluster
static MiniDFSCluster dfsCluster

def setupSpec() {
	def conf = new JobConf()
	if (System.getProperty("hadoop.log.dir") == null) {
		System.setProperty("hadoop.log.dir", "/tmp");
	}
		
	dfsCluster = new MiniDFSCluster(conf, 2, true, null)
	mrCluster = new MiniMRCluster(2, dfsCluster.getFileSystem().getUri().toString(), 1)
		
	def hdfs = dfsCluster.getFileSystem()
	hdfs.delete(new Path('/main-testdata'), true)
	hdfs.delete(new Path('/user'), true)
	FileUtil.copy(inputData, hdfs, new Path('main-testdata'), false, conf)
}

Notes:

• setupSpec() is the Spock equivalent of JUnit 4's @BeforeClass
• If the system property (lines 6 and 7) is not set, the m/r cluster will not startup
• The clusters are configured to use 2 slave nodes each
• After successful filesystem startup, it is possible to perform the usual filesystem operations with it

Teardown and cleanup is similarly performed in the static context:

def cleanupSpec() {
	mrCluster?.shutdown()
	dfsCluster?.getFileSystem().delete(new Path('/main-testdata'), true)
	dfsCluster?.getFileSystem().delete(new Path('/user'), true)
	dfsCluster?.shutdown()
}

However, there is a really annoying problem with this code: it takes forever. I don't know why (presumably, I'm doing something wrong), but this shutdown procedure is blocked by several data integrity tests which take a long time themselves. Since the generated data is garbage and of zero relevance after the tests have completed, I'd like to get rid of these checks, but I really cannot figure out how. The logs get s-l-o-w-l-y filled with lines like

11/09/19 23:47:06 INFO datanode.DataBlockScanner: Verification succeeded for blk_3329401068442722923_1001
11/09/19 23:47:12 INFO datanode.DataBlockScanner: Verification succeeded for blk_654270692326292497_1008
11/09/19 23:47:39 INFO datanode.DataBlockScanner: Verification succeeded for blk_-3673127094860948561_1006

and the single test and thus the whole test suite as well takes several minutes to complete, fatal for each continuous integration system. However, there is a workaround which is neither obvious nor very nice, but it works: wrapping the shutdown procedure in another thread. Simply by modifying the code from above into

def cleanupSpec() {
	Thread.start { 
		mrCluster?.shutdown()
		dfsCluster?.getFileSystem().delete(new Path('/main-testdata'), true)
		dfsCluster?.getFileSystem().delete(new Path('/user'), true)
		dfsCluster?.shutdown()
	}.join(5000)
}

I could not spot any blocking behavior any longer. If someone has a better idea to avoid blocking on shutdown, please feel free to comment.