For our sample job, we'll take all too boring "word count" example. It would be best if we could code it in such a way to satisfy few requirements:
- it should be triggered in the same way from any of 3 target environments
- it should have externalized job parameters (such as "word count" input and output HDFS paths)
- it should have externalized Hadoop configuration to be able to experiment with few of those
- it should be able to take standard hadoop/yarn CLI command parameters
- it should be able to take single input path parameter provided in form of multiple comma-separated paths for cases when Oozie coordinator takes multiple dataset instances during workflow submission
- it should set custom mapreduce job name to improve job visibility on some GUIs (such as Hue)
Taking all these intro consideration, we decided to use recommended way to develop job "driver" applications for hadoop/yarn CLI commands - and that is using Hadoop Tool interface. That way it would be able to parse all standard parameters that these CLI commands provide, such as specifying different configuration file for job submission.
Because Tool application is plain java application anyway, it can be also called from within an IDE. And finally, Oozie can also include it into its workflow as "java action node".
public class WordCount extends Configured implements Tool {
public static void main(String[] args) throws Exception {
int exitCode = ToolRunner.run(new WordCount(), args);
System.exit(exitCode);
}
@Override
public int run(String[] args) throws Exception {
String inputPath = args[0];
String outputPath = args[1];
Configuration conf = getConf();
doWorkflow(inputPath, outputPath, conf);
return 0;
}
...
}
Job parameters - input and output path in this case, are provided as standard java program arguments. But as previously mentioned, when using Oozie, frequently an input argument is single String value containing comma-separated HDFS paths, because Oozie coordinator can be instructed to take multiple instances of some dataset and process them in batch, and it basically generates such comma-separated String value as input argument for triggered Oozie workflow. So its useful to construct source Tap from such String value. So here it goes:
private Tap constructMultiHfsSourceTap(Scheme scheme, String inputPath) {
List<Tap> tapList = new ArrayList<Tap>();
String[] splits = inputPath.split(",");
for (String split : splits) {
tapList.add(new Hfs(scheme, split.trim()));
}
Tap[] taps = tapList.toArray(new Tap[tapList.size()]);
return new MultiSourceTap(taps);
}
We couldn't use Cascading's GlobHfs here since individual paths are not part of some hierarchical structure.
Once this job hits the road, it would be great to easily see few main pieces of information about it on some kind of job UI, so we'll set its mapreduce job name:
....
Flow flow = flowConnector.connect(flowDef);
// set mapreduce job name
String mapReduceJobName = "Cascading Word Count: '" + inputPath + "' -> '" + outputPath + "'";
FlowStepStrategy flowStepStrategy = constructMapReduceJobNameStrategy(mapReduceJobName);
flow.setFlowStepStrategy(flowStepStrategy);
flow.complete();
}
private FlowStepStrategy constructMapReduceJobNameStrategy(final String mapReduceJobName) {
return new FlowStepStrategy() {
@Override
public void apply(Flow flow, List predecessorSteps, FlowStep flowStep) {
Object config = flowStep.getConfig();
if (config instanceof JobConf) {
((JobConf) config).setJobName(mapReduceJobName);
}
}
};
}
Complete sample "word count" project is avaiable on GitHub.
Job submission
IDE
Ok, now when we have all the code in place, we can finally run it locally within our IDE simply by calling the job driver application as any other java application. IDE will take care to put all necessary classes/jars on classpath. java -classpath <complete classpath here set by IDE> vmarcinko.cascading.wordcount.WordCount -conf conf/local-site.xml /home/vmarcinko/cascadingtest/sherlock_holmes.txt /home/vmarcinko/cascadingtest/wordcount
As can be seen, we provided Hadoop configuration file (local-site.xml) by using "-conf" program argument which is standard argument parsed by ToolRunner utility class. In other words, we can use same standard arguments that can be used when submitting job via CLI as shown next.
Sample sherlock_holmes.txt file used in this example is available at <project dir>/data directory. If everything went good (and it should!), then word counts are found in part-xxxx file under output directory.
Command Line Interface (CLI)
First, we must package our job application in suitable form using Gradle Shadow plugin as described in part 1 of this post. The end result of "gradle shadowJar" task would be:<cascading-wordcount project dir>/build/libs/cascading-wordcount-1.0-all.jar
Next we upload that JAR file into Hadoop cluster, place sample .txt file in HDFS path of our choice, and finally submit the job using shell command:
yarn jar cascading-wordcount-1.0-all.jar vmarcinko.cascading.wordcount.WordCount /user/cloudera/cascadingtest/sherlock_holmes.txt /user/cloudera/cascadingtest/wordcount
Shell command "yarn" is available in newer versions of Hadoop. Older version used "hadoop" command.
Oozie
To invoke the same vmarcinko.cascading.wordcount.WordCount application from Oozie, we need to use "java action node" within our Oozie workflow to launch it.Anyway, we use the same shadow JAR (cascading-wordcount-1.0-all.jar) and place it under <oozie workflow HDFS dir>/lib directory. Under program arguments, it would be best to parametrize this java action node with ${inputPath} and ${outputPath}, so we can provide concrete HDFS paths when submitting the workflow.
When the job is launched via Oozie (either manually submitting workflow, or in scheduled manner via Oozie coordinator), we can see our running job nicely in some UI (such as Hue Job Browser in this example): Name of job corresponds to mapreduce job name that we set prior to execution.
(as usual when java application is called via Oozie's "java action node", for each such launch, Oozie initially starts a Map task that acts as launcher for specified java application, thus we end up with 2 jobs shown above)
I hope this post proves useful to all newbies trying to find some common way to set up Cascading job applications that can be triggered from various environments.
