Wednesday, 6 August 2014

MySql vs MongoDB performance benchmark

Problem

We’re searching for the fastest solution to solve the following two use cases:

fastest data retrieval to get 5000/50000 points out of 10 million points,
fastest data retrieval to get 5000/50000/500000 points out of 200 million points.

We use these points to generate JS charts.
For testing choose MongoDB vs MySql. There is no complex relation between tables/objects, the test case is very simple.
Beside the select time we also monitored the fetch time as we need the data to be fetched. We also tested with and without cache (where possible) – our specific project has the possibility to cache data. On this benchmark not tested scalability in detail but it seems that running the test on 8-20 threads the situation doesn’t change very much.

System and config

The benchmark machine is Linux (Ubuntu 12.04 64-bit) with 4 Cores and 10GB of RAM.
MySql cache config (in my.cnf):
query_cache_limit = 1024M
query_cache_type = 1
query_cache_size = 1512M

On Mongo used the default config with journal = true (because didn’t notice any considerably difference on runtime with journal = false and went for the safer option)

Test results

I) Retrieval of 5000/50000 out of 10 million points – lower database load

Generated a random database with 10 million rows. Every row has 5 fields: 2 integers and 3 random strings.
The insert operation is made on 8 threads, inserts are split among threads.
Also for select and fetch used threads (1,2,4 and 8 threads per run) but in our test case every thread runs the same operation.
On select and fetch selected the rows (5000 and 50000 rows) from the total of 10 million rows and made a similar fetch operation for both MySql and MongoDB.
The results output is streamed from shell scripts into csv.

5000 rows selected out of 10 million rows

database	operation	total rows	threads	no of selected rows	run time (select + fetch)	CPU usage (4 cores)	select time
Mongo	Insert	10000000	8	0	411121 ms	137%CPU	0
MySQL	Insert	10000000	8	0	1130493 ms	149%CPU	0
Mongo	Select+Fetch	10000000	1	5000	17411 ms	105%CPU	3 ms	first run
MySQL	Select+Fetch	10000000	1	5000	5836 ms	109%CPU	5369 ms	first run
Mongo	Select+Fetch	10000000	1	5000	6450 ms	116%CPU	2 ms
MySQL	Select+Fetch	10000000	1	5000	512 ms	208%CPU	66 ms
Mongo	Select+Fetch	10000000	2	5000	12507 ms	110%CPU	3 ms
MySQL	Select+Fetch	10000000	2	5000	565 ms	236%CPU	69 ms
Mongo	Select+Fetch	10000000	4	5000	28129 ms	106%CPU	2 ms
MySQL	Select+Fetch	10000000	4	5000	592 ms	255%CPU	72 ms
Mongo	Select+Fetch	10000000	8	5000	75047 ms	64%CPU	1 ms
MySQL	Select+Fetch	10000000	8	5000	759 ms	275%CPU	66 ms

50000 rows selected out of 10 million rows

database	operation	total rows	threads	no of selected rows	run time (select + fetch)	CPU usage (4 cores)	select time
Mongo	Insert	10000000	8	0	410866 ms	137%CPU	0
MySQL	Insert	10000000	8	0	1150706 ms	147%CPU	0
Mongo	Select+Fetch	10000000	1	500000	13049 ms	118%CPU	1 ms	first run
MySQL	Select+Fetch	10000000	1	500000	6128 ms	117%CPU	5599 ms	first run
Mongo	Select+Fetch	10000000	1	500000	6955 ms	138%CPU	2 ms
MySQL	Select+Fetch	10000000	1	500000	959 ms	212%CPU	447 ms
Mongo	Select+Fetch	10000000	2	500000	12445 ms	142%CPU	3 ms
MySQL	Select+Fetch	10000000	2	500000	1110 ms	264%CPU	533 ms
Mongo	Select+Fetch	10000000	4	500000	30251 ms	133%CPU	2 ms
MySQL	Select+Fetch	10000000	4	500000	1483 ms	307%CPU	676 ms
Mongo	Select+Fetch	10000000	8	500000	82642 ms	99%CPU	3 ms
MySQL	Select+Fetch	10000000	8	500000	2220 ms	346%CPU	1433 ms

On operation Select + Fetch on multiple threads ran the same operation for each thread. Ex: on 8 threads made 8 operations of select + fetch (one on each thread).
threads – represents the thread number

CPU usage – represents the CPU usage that the Java process took
first run – it’s the first run without any cache made, the other runs are with cache
run time – it’s the total runtime of select+fetch
select time – it’s the average time of select

Conclusions for this test case

On first run (no cache on MySql): total fetch + select Mongo time is 17.4s (3ms select time) at MySql it is 5.8s (5.3s select time). Notice here the Mongo select time 3 ms! This is probably caused by the fact that it has some lazy processing.

On next runs (the MySql cache enters): total fetch + select Mongo time is 6.9s (2ms select time) at MySql it is 0.9s (0.4s select time). Notice here the big impact of cache on MySql!

For a project of this scale that also has the possibility to cache the data requests it seems that overall MySql is much better for read operations.
Even If not possible to cache the data MySql would still be a winner. MySql better by a 2X – 5X speed for reading.
Notice the good insert speed on Mongo: it is 2-3X faster than MySql.

The 50000 test gives similar results.

II) Retrieval of 5000/50000/500000 out of 200 million points – higher database load

Generated a random database with 200 million rows. Every row has 5 fields: 2 integers and 3 random strings.
The insert operation is made on 8 threads, inserts are split among threads.
Also for select and fetch used threads (1,2,4 and 8 threads per run) but in our test case every thread runs the same operation.
On select and fetch selected the rows (5000, 50000 and 500000 rows) from the total of 200 million rows and made a similar fetch operation for both MySql and Mongo.
The results output is streamed from some shell scripts into csv.

5000 selected rows out of 200 million rows

database	operation	total rows	threads	no of selected rows	run time (select + fetch)	CPU usage (4 cores)	select time
Mongo	Insert	200000000	8	0	8283056 ms	134%CPU	0
MySQL	Insert	200000000	8	0	22270948 ms	147%CPU	0
Mongo	Select+Fetch	200000000	1	5000	185969 ms	100%CPU	4 ms	first run
MySQL	Select+Fetch	200000000	1	5000	325609 ms	100%CPU	324876 ms	first run
Mongo	Select+Fetch	200000000	1	5000	170406 ms	100%CPU	0 ms
MySQL	Select+Fetch	200000000	1	5000	797 ms	131%CPU	64 ms
Mongo	Select+Fetch	200000000	2	5000	366705 ms	100%CPU	3 ms
MySQL	Select+Fetch	200000000	2	5000	1088 ms	151%CPU	71 ms
Mongo	Select+Fetch	200000000	4	5000	626023 ms	100%CPU	3 ms
MySQL	Select+Fetch	200000000	4	5000	846 ms	158%CPU	85 ms
Mongo	Select+Fetch	200000000	8	5000	1564526 ms	73%CPU	2 ms
MySQL	Select+Fetch	200000000	8	5000	1048 ms	181%CPU	38 ms

50000 selected rows out of 200 million rows

database	operation	total rows	threads	no of selected rows	run time (select + fetch)	CPU usage (4 cores)	select time
Mongo	Insert	200000000	8	0	8366874 ms	134%CPU	0
MySQL	Insert	200000000	8	0	22592623 ms	148%CPU	0
Mongo	Select+Fetch	200000000	1	50000	191197 ms	100%CPU	2 ms	first run
MySQL	Select+Fetch	200000000	1	50000	323260 ms	100%CPU	322279 ms	first run
Mongo	Select+Fetch	200000000	1	50000	172113 ms	100%CPU	3 ms
MySQL	Select+Fetch	200000000	1	50000	1374 ms	120%CPU	144 ms
Mongo	Select+Fetch	200000000	2	50000	349728 ms	100%CPU	3 ms
MySQL	Select+Fetch	200000000	2	50000	991 ms	154%CPU	113 ms
Mongo	Select+Fetch	200000000	4	50000	596883 ms	100%CPU	2 ms
MySQL	Select+Fetch	200000000	4	50000	1160 ms	180%CPU	134 ms
Mongo	Select+Fetch	200000000	8	50000	1446121 ms	75%CPU	2 ms
MySQL	Select+Fetch	200000000	8	50000	1296 ms	201%CPU	196 ms

500000 selected rows out of 200 million rows

database	operation	total rows	threads	no of selected rows	run time (select + fetch)	CPU usage (4 cores)	select time
Mongo	Insert	200000000	8	0	8388447 ms	134%CPU	0
MySQL	Insert	200000000	8	0	23010922 ms	148%CPU	0
Mongo	Select+Fetch	200000000	1	500000	188207 ms	101%CPU	2 ms	first run
MySQL	Select+Fetch	200000000	1	500000	321247 ms	100%CPU	320165 ms	first run
Mongo	Select+Fetch	200000000	1	500000	172975 ms	101%CPU	2 ms
MySQL	Select+Fetch	200000000	1	500000	1265 ms	162%CPU	460 ms
Mongo	Select+Fetch	200000000	2	500000	371833 ms	103%CPU	3 ms
MySQL	Select+Fetch	200000000	2	500000	1514 ms	188%CPU	511 ms
Mongo	Select+Fetch	200000000	4	500000	626729 ms	104%CPU	0 ms
MySQL	Select+Fetch	200000000	4	500000	1768 ms	230%CPU	711 ms
Mongo	Select+Fetch	200000000	8	500000	1545454 ms	78%CPU	2 ms
MySQL	Select+Fetch	200000000	8	500000	2403 ms	293%CPU	1374 ms

Conclusions for this test case

On first run (no cache on MySql): total fetch + select Mongo time is 85s (4ms select time) at MySql it is 325s (324s select time). Again the select at Mongo is fast: 4 ms. Probably at select Mongo makes nothing, the lazy processing implementation appears again and the fetch is slower.

On next runs (the MySql cache enters): total fetch + select Mongo time is 170s (0ms select time) at MySql it is 0.7s (0.064s select time). Notice here the big impact of cache on MySql!

For a project of this scale that also has the possibility to cache the data requests it seems that overall MySql is still much better for read operations.
If it is not possible to cache data Mongo would be almost 1.3X – 2X faster than MySql.
Notice the good insert speed on Mongo: it is 2-3X faster than MySql.

The 50000 and 500000 tests give similar results.

Source code used for tests

MySQL Code:
database drop and create:

Class.forName(driver).newInstance();
con = DriverManager.getConnection(url + db, user, pass);
con.createStatement().execute("DROP TABLE test");
con.createStatement().execute(
"CREATE TABLE test"
+ "(id INTEGER not NULL AUTO_INCREMENT, "
+ " test1 VARCHAR(255), " + " test2 VARCHAR(255), "
+ " test3 VARCHAR(255), " + " test4 INTEGER, "
+ " PRIMARY KEY ( id ))");
con.close();

Class.forName(driver).newInstance();

con = DriverManager.getConnection(url + db, user, pass);

con.createStatement().execute("DROP TABLE test");

con.createStatement().execute(

"CREATE TABLE test"

+ "(id INTEGER not NULL AUTO_INCREMENT, "

+ " test1 VARCHAR(255), " + " test2 VARCHAR(255), "

+ " test3 VARCHAR(255), " + " test4 INTEGER, "

+ " PRIMARY KEY ( id ))");

con.close();

database insert code executed on 8 threads:

mysqlgen3.argu – rows number (example 10 million) / threads (example 8 threads)
mysqlgen3.nrselect – it`s the number of the selected rows

Class.forName(driver).newInstance();
con = DriverManager.getConnection(url + db, user, pass);
final Statement st = con.createStatement();
String c, d, e;
int x = 0;
for (i = 0; i < mysqlgen3.argu; i++) {
if (i % 250000 == 0) {
System.gc();
}
final String randomString = UUID.randomUUID().toString();
c = randomString.substring(0, 5);
d = randomString.substring(5, 10);
e = randomString.substring(10, 15);
if (mysqlgen3.nrselect >= nrcount) {
nrcount++;
st.execute("INSERT INTO `test` (`test1`, `test2`, `test3`, `test4`) VALUES ('"
+ c + "', '" + d + "', '" + e + "', 12345)");
} else {
x = generator.nextInt(9999);
st.execute("INSERT INTO `test` (`test1`, `test2`, `test3`, `test4`) VALUES ('"
+ c + "', '" + d + "', '" + e + "', '" + x + "')");
}
}
con.close();

Class.forName(driver).newInstance();

con = DriverManager.getConnection(url + db, user, pass);

final Statement st = con.createStatement();

String c, d, e;

int x = 0;

for (i = 0; i < mysqlgen3.argu; i++) {

if (i % 250000 == 0) {

System.gc();

}

final String randomString = UUID.randomUUID().toString();

c = randomString.substring(0, 5);

d = randomString.substring(5, 10);

e = randomString.substring(10, 15);

if (mysqlgen3.nrselect >= nrcount) {

nrcount++;

st.execute("INSERT INTO `test` (`test1`, `test2`, `test3`, `test4`) VALUES ('"

+ c + "', '" + d + "', '" + e + "', 12345)");

} else {

x = generator.nextInt(9999);

st.execute("INSERT INTO `test` (`test1`, `test2`, `test3`, `test4`) VALUES ('"

+ c + "', '" + d + "', '" + e + "', '" + x + "')");

}

con.close();

select + fetch:

ResultSet res;
long startTime, endTime, totalTime;
Class.forName(driver).newInstance();
con = DriverManager.getConnection(url + db, user, pass);
Statement st = con.createStatement();
startTime = System.currentTimeMillis();
try {
res = st.executeQuery("SELECT * FROM test WHERE `test4`=12345");
totalTime = System.currentTimeMillis() - startTime;
System.out.println("Select time: " + totalTime + " ms");
startTime = System.currentTimeMillis();
while (res.next()) {
count++;
}
totalTime = System.currentTimeMillis() - startTime;
System.out.println("Fetch time: " + totalTime + " ms");
System.out.println("Rows: " + count);</p>
<b>Mongo Code:</b>
<b>database drop and create:</b>
<p align="LEFT">
DBCollection collection;
Mongo mongo = new Mongo("localhost", 27017);
DB db = mongo.getDB("local");
collection = db.getCollection("test");
collection.drop();
if (collection == null) {
collection = db.createCollection("test", null);
}
mongo.close();

ResultSet res;

long startTime, endTime, totalTime;

Class.forName(driver).newInstance();

con = DriverManager.getConnection(url + db, user, pass);

Statement st = con.createStatement();

startTime = System.currentTimeMillis();

try {

res = st.executeQuery("SELECT * FROM test WHERE `test4`=12345");

totalTime = System.currentTimeMillis() - startTime;

System.out.println("Select time: " + totalTime + " ms");

startTime = System.currentTimeMillis();

while (res.next()) {

count++;

}

totalTime = System.currentTimeMillis() - startTime;

System.out.println("Fetch time: " + totalTime + " ms");

System.out.println("Rows: " + count);</p>

<strong>Mongo Code:</strong>

<strong>database drop and create:</strong>

<p align="LEFT">DBCollection collection;

Mongo mongo = new Mongo("localhost", 27017);

DB db = mongo.getDB("local");

collection = db.getCollection("test");

collection.drop();

if (collection == null) {

collection = db.createCollection("test", null);

}

mongo.close();

database insert code executed on 8 threads:

DBCollection collection;
Mongo mongo = new Mongo("localhost", 27017);
DB db = mongo.getDB("local");
collection = db.getCollection("test");
String c, d, e;
int x = 0;
BasicDBObject document;
for (i = 0; i < mongogen3.argu; i++) {
if (i % 10000 == 0) {
System.gc();
}
final String randomString = UUID.randomUUID().toString();
c = randomString.substring(0, 5);
d = randomString.substring(5, 10);
e = randomString.substring(10, 15);
if (mongogen3.nrselect >= nrcount) {
nrcount++;
document = new BasicDBObject();
document.put("test1", "" + c + "");
document.put("test2", "" + d + "");
document.put("test3", "" + e + "");
document.put("test4", 12345);
collection.insert(document);
} else {
x = generator.nextInt(9999);
document = new BasicDBObject();
document.put("test1", "" + c + "");
document.put("test2", "" + d + "");
document.put("test3", "" + e + "");
document.put("test4", "" + x + "");
collection.insert(document);
}
}
mongo.close();

DBCollection collection;

Mongo mongo = new Mongo("localhost", 27017);

DB db = mongo.getDB("local");

collection = db.getCollection("test");

String c, d, e;

int x = 0;

BasicDBObject document;

for (i = 0; i < mongogen3.argu; i++) {

if (i % 10000 == 0) {

System.gc();

}

final String randomString = UUID.randomUUID().toString();

c = randomString.substring(0, 5);

d = randomString.substring(5, 10);

e = randomString.substring(10, 15);

if (mongogen3.nrselect >= nrcount) {

nrcount++;

document = new BasicDBObject();

document.put("test1", "" + c + "");

document.put("test2", "" + d + "");

document.put("test3", "" + e + "");

document.put("test4", 12345);

collection.insert(document);

} else {

x = generator.nextInt(9999);

document = new BasicDBObject();

document.put("test1", "" + c + "");

document.put("test2", "" + d + "");

document.put("test3", "" + e + "");

document.put("test4", "" + x + "");

collection.insert(document);

}

mongo.close();

select + fetch:

long startTime, totalTime;
Mongo mongo = new Mongo("localhost", 27017);
DB db = mongo.getDB("local");
DBCollection collection = db.getCollection("test");
BasicDBObject query;
DBCursor cursor;
query = new BasicDBObject();
query.put("test4", 12345);
startTime = System.currentTimeMillis();
cursor = collection.find(query);
totalTime = System.currentTimeMillis() - startTime;
System.out.println("Select time: " + totalTime + " ms");
startTime = System.currentTimeMillis();
while (cursor.hasNext()) {
count++;
cursor.next();
}
totalTime = System.currentTimeMillis() - startTime;
System.out.println("Fetch time: " + totalTime + " ms");
System.out.println("Rows: " + count);

long startTime, totalTime;

Mongo mongo = new Mongo("localhost", 27017);

DB db = mongo.getDB("local");

DBCollection collection = db.getCollection("test");

BasicDBObject query;

DBCursor cursor;

query = new BasicDBObject();

query.put("test4", 12345);

startTime = System.currentTimeMillis();

cursor = collection.find(query);

totalTime = System.currentTimeMillis() - startTime;

System.out.println("Select time: " + totalTime + " ms");

startTime = System.currentTimeMillis();

while (cursor.hasNext()) {

count++;

cursor.next();

}

totalTime = System.currentTimeMillis() - startTime;

System.out.println("Fetch time: " + totalTime + " ms");

System.out.println("Rows: " + count);

On both MySQL and Mongo used threads for select+fetch and insert. Example of implemented threads:
threadCount is the number of threads and MyThread class contains the presented code above.

for (int x = 0; x < threadCount; x++) {
MyThread temp = new MyThread("Thread #" + x);
threads.add(temp);
temp.start();
System.out.println("Started Thread:" + x);
}
boolean allDone = true;
while (allDone != false) {
allDone = false;
for (int x = 0; x < threadCount; x++) {
allDone |= threads.get(x).isAlive();
}
}

for (int x = 0; x < threadCount; x++) {

MyThread temp = new MyThread("Thread #" + x);

threads.add(temp);

temp.start();

System.out.println("Started Thread:" + x);

}

boolean allDone = true;

while (allDone != false) {

allDone = false;

for (int x = 0; x < threadCount; x++) {

allDone |= threads.get(x).isAlive();

}

Saturday, 2 August 2014

Scheduling Workflows Using Oozie Coordinator

Introduction

In Hadoop Ecosystem, most of the functionality like map-reduce jobs, pig scripts and hive queries are executed as batch jobs. This creates a lot of overhead in deployment and maintenance of hadoop components. As a solution to this, Oozie provides workflows in xml format using which we can define multiple Map/Reduce jobs into a logical unit of work, accomplishing the larger task [4]. This helps in chaining the related MapReduce jobs which can be either Hive queries or Pig scripts like mentioned in below diagram.

Workflows work perfectly when invoked on demand or manually. But for achieving higher level of automation and effectiveness, it becomes necessary to run them based on one or more of the following parameters: regular time intervals, data availability or external events. Then, we need more functionality than provided by Oozie workflows.

In this paper, Oozie Coordinator Jobs will be discussed which provide options to embed workflows and trigger them on regular time-intervals or on basis of data availability.

The Oozie coordinator allows expressing conditions to trigger execution of workflow in the form of the predicates [1]. These predicates are conditional statements on parameters like time, data and external events. If the predicate is satisfied, then only the workflow job/action is started.

Oozie Coordinator System

As stated at Oozie documentation page [1], “Oozie is a Java Web-Application that runs in a Java servlet-container”. It uses XML for taking configuration inputs from user and uses a database (default is derby but MySQL, HSQLDB or any RDBMS database can also be used) to store:

Definitions of Workflow and Coordinator
Currently running workflow and Coordinator instances, including instance states, configuration variables and parameters.

Oozie Coordinator is a collection of predicates (conditional statements based on time-frequency and data availability) and actions (i.e. Hadoop Map/Reduce jobs, Hadoop file system, Hadoop Streaming, Pig, Java and Oozie sub-workflow). Actions are recurrent workflow jobs invoked each time predicate returns true.

Oozie version 2 and higher supports Coordinator Jobs. Coordinator Job is defined in the XML Process Definition Language.

Predicates are conditional statements, defined using attributes “interval, start-time and end-time” for time-based triggering and xml-tags “dataset and input-events” for data-availability based triggering of workflows.

Actions are the mechanism by which a workflow is triggered for the execution of a computation/processing task. Action contains description of one or more workflows to be executed.

Oozie is lightweight as it uses existing Hadoop Map/Reduce framework for executing all tasks in a workflow. This approach allows it to leverage existing Hadoop installation for providing scalability, reliability, parallelism, etc.

On the basis of functionality, Coordinator can be sub-divided into two major groups [2]:

1. Time-Based Coordinator: This type of Coordinator definition is used for invoking the workflow repeatedly after an interval between a specified period of time.

2.File-Based Coordinator: This type of Coordinator definition is used for invoking the workflow on the basis of data availability and data polling.

2.1 Simple File-Based Coordinator: The action is invoked whenever data available predicate is true.

2.2 Sliding Window-Based Coordinator: It is invoked frequently and data is aggregated over multiple overlapping previous instances. For example, invoking it at a frequency of 5 minutes and running action on aggregated previous 4 instances of 15 minutes data.

2.3Rollups-Based Coordinator: It is invoked after a long period of time and data is aggregated over multiple previous instances from last time of invocation. For example, it will run once a day, and will trigger a workflow that aggregates 24 instances of hourly data.

Oozie Coordinator Components and Variables

Coordinator-App: It is a wrapper component that defines the attributes of a coordinator and includes all other components.

Attributes are:

start , end : describes the start and end time in yyyy-mm-ddThh:mmZ format
Time zone: describes the time zone (is the value of Z in the above time format) like UTC.
Controls: It contains parameters like timeout, concurrency, etc. to configure the execution of coordinator job.
Datasets: It contains the definition of multiple data sources and frequency of data polling.

Attributes are:

Frequency: interval of time at which data polling is done.
Initial-Instance: start time of data polling in yyyy-mm-ddThh:mmZ format.
Uri-Template: URI of the data source. Expression language can be used. For example, ${YEAR} corresponds to current year. It helps in dynamic selection of data source directories.
Done-flag: This flag denotes the success of data polling. It can be a file in which case the presence of file is checked before calling action. It can be left empty otherwise for implicit success message.
Input-Events: denotes the processing of the input data before running the action.
Data-in: it denotes the aggregated output data of input-event.
Start-instance and end-instance: boundary of data instances that needs to be aggregated.
Output-Events: denotes the processing of the output data after running the action.
Data-out: it denotes the output dataset.
Instance: instance of dataset that is to be used as sink for output.
Action: It includes the path of the workflow that has to be invoked when predicate return true.

It could also be configured to record the events required to evaluate SLA compliance.

Oozie Coordinator Lifecycle Operations

The lifecycle operations of coordinator are similar to those of oozie workflow except start operation. “Start” is not applicable for coordinators.

Submit/Run: Both operations submit the coordinator job to oozie. The job will be in PREP state till the mentioned start-time of the coordinator.
Suspend: Suspends/pause the coordinator job.
Resume: Resumes the execution of the coordinator job.
Kill: kill the coordinator job and ends its execution.
reRun: re-submitting the coordinator job/actions with new parameters.

Oozie Coordinator Example

In this section, we will see how to use oozie coordinator for scheduling and triggering of the workflows.

A Sample Workflow: First of all, we need a oozie workflow job. For example purpose, I have taken the simple wordcount example provided by Apache-Hadoop-Distribution in hadoop-examples-0.20.2-cdh3u0.jar [6].

The workflow for wordcount is:

<workflow-app xmlns='uri:oozie:workflow:0.1' name='java-main-wf'>
<start to='mapreduce-wordcount-example' />
<action name='mapreduce-wordcount-example'>
<java>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
<configuration>
<property>
<name>mapred.job.queue.name</name>
<value>default</value>
</property>
</configuration>
<main-class>org.apache.hadoop.examples.ExampleDriver</main-class>
<arg>wordcount</arg>
<arg>${inputDir}</arg>
<arg>${outputDir}</arg>
</java>
<ok to="end" />
<error to="fail" />
</action>
<kill name="fail">
<message>Java failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
</kill>
<end name='end' />
</workflow-app>

Once workflow is created it has to be deployed correctly. A typical Oozie deployment is a HDFS directory, containing workflow.xml and a lib subdirectory, containing jar files of classes used by workflow actions.

For example, the directory structure in hadoop will be as shown below. (If user.name is training)

[hadoop@localhost ~]$ hadoop dfs -ls /user/training/oozie/workflow/wordcount
Found 2 items
drwxr-xr-x - training supergroup 0 2012-09-18 12:05 /user/training/oozie/workflow/wordcount/lib
-rw-r--r-- 1 training supergroup 918 2012-09-18 11:47 /user/training/oozie/workflow/wordcount/workflow.xml

The job.properties file will have following properties:

nameNode=hdfs://localhost:8020
jobTracker=localhost:8021
queueName=default
inputDir=${nameNode}/data.in
outputDir=${nameNode}/out
user.name=training
oozie.wf.application.path=${nameNode}/user/${user.name}/oozie/workflow/wordcount/

With job properties in place, this workflow can be invoked manually using the oozie workflows submit command from command-line.

[training@localhost Desktop]$ oozie job -oozie=http://localhost:11000/oozie/ -config oozie/wordcount-demo/workflow/job.properties -run;

job: 0000000-120918134457517-oozie-oozi-W

2. Oozie Coordinator Definition: As discussed above, coordinator-definitions will be different for different kind of triggering and scheduling.

So, we will take each kind of Coordinator one by one and schedule wordcount example on the basis of that.

Moreover, Oozie coordinators can be parameterized using variables like ${inputDir}, ${startTime}, etc. within the coordinator definition. When submitting a coordinator job, values for the parameters must be provided as input. As parameters are key-value pairs, they can be written in a job.properties file or a XML file. Parameters can also be provided in form of a java Map object if using JAVA API to invoke a coordinator job.

Time-Based Coordinator

The generic definition for this kind of coordinator is

<coordinator-app name="coordinator1" frequency="${frequency}" start="${startTime}" end="${endTime}" timezone="${timezone}" xmlns="uri:oozie:coordinator:0.1">
<action>
<workflow>
<app-path>${workflowPath}</app-path>
</workflow>
</action>
</coordinator-app>

Save the file as coordinator.xml in a HDFS directory. (Please note that coordinator.xml is the only name which can be given to the file as oozie uses this default name for reading file in HDFS directory.)

The coordinatorjob.properties can be defined as
frequency=60
startTime=2012-08-31T20\:20Z
endTime=2013-08-31T20\:20Z
timezone=GMT+0530
workflowPath=${nameNode}/user/${user.name}/oozie/workflow/wordcount/
nameNode=hdfs://localhost:8020
jobTracker=localhost:8021
queueName=default
inputDir=${nameNode}/data.in
outputDir=${nameNode}/out

oozie.coord.application.path=${nameNode}/user/${user.name}/coordOozie/coordinatorTimrBased

The coordinator application path must be specified in the file with the oozie.coord.application.path property. Specified path must be an HDFS path.

File-Based Coordinator

<coordinator-app name="coordinator1" frequency="${frequency}" start="${startTime}" end="${endTime}" timezone="UTC" xmlns="uri:oozie:coordinator:0.1">
<datasets>
<dataset name="input1" frequency="${datasetfrequency}" initial-instance="${datasetinitialinstance}"
timezone="${datasettimezone}">
<uri-template>${dataseturitemplate}/${YEAR}/${MONTH}/${DAY}/${HOUR}/
${MINUTE}</uri-template>
<done-flag> </done-flag>
</dataset>
</datasets>
<input-events>
<data-in name="coordInput1" dataset="input1">
<start-instance>${inputeventstartinstance}</start-instance>
<end-instance>${inputeventendinstance}</end-instance>
</data-in>
</input-events>
<action>
<workflow>
<app-path>${workflowPath}</app-path>
</workflow>
</action>
</coordinator-app>

The coordinatorjob.properties can be defined as

frequency=60
startTime=2012-08-21T15:25Z
endTime=2012-08-22T15:25Z
timezone=UTC
datasetfrequency=15
datasetinitialinstance=2012-08-21T15:30Z
datasettimezone=UTC
dataseturitemplate=${namenode}/user/hadoop/oozie/coordinator/in
inputeventstartinstance=${coord:current(0)}
inputeventendinstance=${coord:current(0)}
workflowPath=${nameNode}/user/${user.name}/oozie/workflow/wordcount/
nameNode=hdfs://localhost:8020
jobTracker=localhost:8021
queueName=default
inputDir= ${coord:dataIn('coordInput1')}
outputDir=${nameNode}/out
oozie.coord.application.path=${nameNode}/user/${user.name}/coordOozie/coordinatorFileBased

The coordinator application path must be specified in the file with the oozie.coord.application.path property. Specified path must be an HDFS path.

Sliding-Window Based Coordinator

This is a specific usecase for the File-Based Coordinator where coordinator is invoked frequently and data is aggregated over multiple overlapping previous instances.

The rule for this can be generalized as
Coordinator-frequency < DataSet-Frequency

For example, the coordinator job.properties will be like

frequency=5
…
datasetfrequency=15
……

Rollups Based Coordinator

This is a specific usecase for the File-Based Coordinator where coordinator is invoked after a long period of time and data is aggregated over multiple previous instances from last time of invocation.

The rule for this can be generalized as
Coordinator-frequency > DataSet-Frequency

frequency=1440
….
datasetfrequency=60
…….

Running Coordinator Example from Command line

Submitting/Running the coordinator job

$ oozie job -oozie http://localhost:11000/oozie -config coordinatorjob.properties [-submit][-run]

job: 0000672-120823182447665-oozie-hado-C

The parameters for the job must be provided in a file, either a Java Properties file (.properties) or a Hadoop XML Configuration file (.xml). This file must be specified with the -config option.

Suspending the coordinator job

$ oozie job -oozie http://localhost:11000/oozie -suspend 0000673-120823182447665-oozie-hado-C

Resuming a Coordinator Job

$ oozie job -oozie http://localhost:11000/oozie -resume 0000673-120823182447665-oozie-hado-C

Killing a Coordinator Job

$ oozie job -oozie http://localhost:11000/oozie -kill 0000673-120823182447665-oozie-hado-C

Rerunning a Coordinator Action or Multiple Actions

$ oozie job -rerun 0000673-120823182447665-oozie-hado-C [-nocleanup]

[-refresh][-action 1,3-5] [-date 2012-01-01T01:00Z::2012-05-31T23:59Z, 2012-11-10T01:00Z, 2012-12-31T22:00Z]

-action or -date is required to rerun. If neither -action nor -date is given, the exception will be thrown.

Checking the Status of a Coordinator/Workflow job or a Coordinator Action

$ oozie job -oozie http://localhost:11000/oozie -info 0000673-20823182447665-oozie-hado-C

The info option can display information about a workflow job or coordinator job or coordinator action.

Invoking Coordinator Jobs from Java Client

The Oozie has exposed a JAVA API for invoking and controlling the workflows programmatically. Same API is also made applicable for coordinator but with some changes as coordinator and workflow differ in functioning.

//The service for executing coordinators on oozie
public class CoordinatorOozieService
{
    // Oozie Client
    OozieClient oozieClient = null;

    public CoordinatorOozieService(String url){
        oozieClient = new OozieClient(url);
    }

    //To submit the coordinator job on oozie
    public String submitJob(String jobPropertyFilePath) throws OozieClientException, IOException{

        // create an empty coordinator job configuration object
        //with just the USER_NAME set to the JVM user name
        Properties conf = oozieClient.createConfiguration();

        conf.setProperty("user.name", "training");

        //set the coordinator properties
        conf.load(new FileInputStream(jobPropertyFilePath));

        // submit the coordinator job
        return oozieClient.submit(conf);
    }

    //To submit the coordinator job on oozie
    public String submitJob(Properties workflowProperties) throws OozieClientException, IOException{

        // create an empty coordinator job configuration object
        //with just the USER_NAME set to the JVM user name
        Properties conf = oozieClient.createConfiguration();

        //set the coordinator properties
        conf.putAll(workflowProperties);

        conf.setProperty("user.name", "training");

        // submit the coordinator job
        return oozieClient.submit(conf);
        }

        // To run (submit and start) the coordinator job on oozie
        public String runJob(String jobPropertyFilePath) throws OozieClientException, IOException{

        // create an empty coordinator job configuration object
        //with just the USER_NAME set to the JVM user name

        Properties conf = oozieClient.createConfiguration();

        conf.setProperty("user.name", "training");

        //set the coordinator properties
        conf.load(new FileInputStream(jobPropertyFilePath));

        // submit and start the coordinator job
        return oozieClient.run(conf);
    }

    // To suspend the coordinator job on oozie
    public void suspendJob(String jobId) throws OozieClientException {
        // start the coordinator job
        oozieClient.suspend(jobId);
    }

    // To resume the coordinator job on oozie
    public void resumeJob(String jobId) throws OozieClientException {
        // start the coordinator job
        oozieClient.resume(jobId);
    }

    //To kill the coordinator job on oozie
    public void killJob(String jobId) throws OozieClientException {
        // start the coordinator job
        oozieClient.kill(jobId);
    }

    //To get the status of the Coordinator Job with id <jobID>
    public Status getJobStatus(String jobID) throws OozieClientException{
        CoordinatorJob job = oozieClient.getCoordJobInfo(jobID);
        return job.getStatus();
    }
}

Conclusion

The Oozie Coordinator can be used for efficient scheduling of the Hadoop-related workflows. It also helps in triggering the same on the basis of availability of the data or external events. Moreover, it provides lot of configurable and pluggable components which helps in easy and effective deployment and maintenance of the Oozie workflow jobs.
As the coordinator is specified in XML, it is easy to integrate it with the J2EE applications. Invoking of coordinator jobs through java has already been explained above.

Enhancements

Oozie provides a new component, “Bundle” in its latest version 3. It provides a higher-level abstraction in which it creates a set of coordinator applications often called a Data Pipeline. Data Dependency can be inserted between multiple coordinator jobs to create an implicit data application pipeline. Oozie Lifecycle operations (start/stop/suspend/resume/rerun) can also be applied at the bundle level which results in a better and easy operational control.

References

[1] Oozie Yahoo! Workflow Engine for Hadoop: http://incubator.apache.org/oozie/docs/3.1.3-incubating/docs/
[2] Oozie Coord Use Cases: https://github.com/yahoo/oozie/wiki/Oozie-Coord-Use-Cases
[3] Better Workflow Management in CDH Using Oozie 2: https://github.com/yahoo/oozie/wiki/Oozie-Coord-Use-Cases

[4] Introduction to Oozie: http://www.infoq.com/articles/introductionOozie

[5] Apache Hadoop: http://hadoop.apache.org/
[6] Index of public/org/apache/hadoop/hadoop-examples/0.20.2-cdh3u0:
https://repository.cloudera.com/artifactory/public/org/apache/hadoop/hadoop-examples/0.20.2-cdh3u0/

Kalyan Hadoop Training in Hyderabad @ ORIEN IT, Ameerpet, 040 65142345 , 9703202345

Pages

Wednesday, 6 August 2014

MySql vs MongoDB performance benchmark

MySql vs MongoDB performance benchmark

Problem

System and config

Test results

Source code used for tests

Saturday, 2 August 2014

Scheduling Workflows Using Oozie Coordinator

Scheduling Workflows Using Oozie Coordinator

Introduction

Oozie Coordinator System

Oozie Coordinator Lifecycle Operations

Oozie Coordinator Example

Running Coordinator Example from Command line

Invoking Coordinator Jobs from Java Client

Conclusion

Enhancements

References