Oozie - Incremental Table Load Workflow

Objective:

Build a oozie workflow that would incrementally load a table from mysql and create a matching table in hive. Schedule this workflow to run every hour.

Launch mysql

$ mysql -u root -p

<cloudera is the password>

mysql> USE retail_db;

mysql> DROP TABLE IF EXISTS employee;

mysql> CREATE TABLE employee(

emp_id INT NOT NULL AUTO_INCREMENT,

name VARCHAR(100),

age INT,

delete_flag TINYINT(1),

LastModifiedDate TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

PRIMARY KEY (emp_id)

);

mysql> INSERT INTO employee (name, age) VALUES

("Employee 1", 34),

("Employee 2", 40);

mysql> SELECT * FROM employee;

Create a .password file containing db user's password. Here the user in mysql db is "root" and corresponding password id "cloudera".

$ echo -n "cloudera" > .password

Upload the .password to /sqoop directory in HDFS.

$ sudo -u yarn hdfs dfs -mkdir /sqoop

$ sudo -u yarn hdfs dfs -put .password /sqoop

Restrict the access only to yarn user.

$ sudo -u yarn hdfs dfs -chmod 400 /sqoop/.password

Remove the .password from the local system for security reason.

$ rm -f .password

Create a sqoop job for incremental load in append mode. Create this job under yarn user because we will ultimately build a oozie job that runs under yarn user.

$ hadoop fs -rm -r -f /staging/mysql/retail_db/employee

$ sudo -u yarn sqoop job \

--create import-employee \

-- import \

--connect jdbc:mysql://localhost:3306/retail_db \

--username root \

--password-file /sqoop/.password \

--table employee \

--target-dir '/staging/mysql/retail_db/employee' \

--incremental lastmodified \

--check-column LastModifiedDate \

--append \

--driver com.mysql.jdbc.Driver \

-m 1

View the existing jobs under yarn user.

$ sudo -u yarn sqoop job --list

You can test the sqoop job by running

$ sudo -u yarn sqoop job --exec import-employee

Once test is successful delete the sqoop job and delete the HDFS target directory.

$ sudo -u yarn sqoop job --delete import-employee

$ hadoop fs -rm -r /staging/mysql/retail_db/employee

And re-create the sqoop job import-employee as earlier.

Verify that the sqoop job is visible as yarn user

$ sudo -u yarn sqoop job --list

Open hive terminal and create the following two tables

hive> insert overwrite table employee

select id, name, age, deleted, lmd from

(select

*,

row_number() over (partition by id order by lmd desc) r

from employee_staging) t

where r = 1;

Select the workflow you created and click on Schedule.