sqoop incremental import in cloudera hadoop

-




In the last blog post, I described how we can import data from RDBMS to HDFS using sqoop.

Now will discuss how we can do incremental import in cloudera hadoop user interface. If you know the basic functionalities on hadoop, this is a simple task!

You need to consider ‘incremental’, ‘check-column’, and ‘last-value’ options to perform the incremental import in sqoop.

Following syntax is using for the incremental import
--incremental <mode>
--check-column <column name>
--last value <last check column value>

Cloudera hadoop is a commercial version of the hadoop. I am using Oozie workflow UI provided by the cloudera to import data.

When you are defining workflows in Oozie UI, you need to give the correct file path for the JDBC driver as well. If you didn’t include the drivers yet, please make sure you include all of those in a folder that can be accessed by everyone.

Login to the Hue UI -> Workflows -> editors -> workflows



I tried to execute below sqoop job in the xml format.


<sqoop xmlns="uri:oozie:sqoop-action:0.2">
  <job-tracker>yarnRM</job-tracker>
  <name-node>hdfs://nameservice1</name-node>
  <arg>import</arg>
  <arg>--connect</arg>
  <arg>jdbc:mysql://172.23.123.132:3306/ideabizAdmin</arg>
  <arg>--username</arg>
  <arg>hadoop</arg>
  <arg>--password</arg>
  <arg>hadoop</arg>
  <arg>--incremental</arg>
  <arg>append</arg>
  <arg>--table</arg>
  <arg>pin_charge_txn</arg>
  <arg>--check-column</arg>
  <arg>DATETIME</arg>
  <arg>--last-value</arg>
  <arg>2017-12-22 23:56:00</arg>
  <arg>--target-dir</arg>
  <arg>/user/thilina_02338/source/ideabiz_admin/pin/charge/data</arg>
  <arg>--as-textfile</arg>
  <arg>--fields-terminated-by</arg>
  <arg>"|"</arg>
  <arg>--null-string</arg>
  <arg>\0</arg>
  <arg>--null-non-string</arg>
  <arg>\0</arg>
  <arg>-m</arg>
  <arg>1</arg>
  <file>/user/oozie/share/lib/lib_20170927010640/sqoop/mysql-connector-java-5.1.38-bin.jar#mysql-connector-java-5.1.38-bin.jar</file>
  <configuration />
</sqoop>

Let’s discuss how we can use incremental update in sqoop job.

In the above job, last value will be the DATETIME column in the table. When you design the job using Oozie and schedule by coordinates, it is capable of taking the last updated value from the system time and to compare with the last DATETIME value in the internal table.

Now there should be a question in your mind, can’t we use any other variables instead of DATETIME. Yes you can. But you should have to find a way to get the last value of your external/internal table and use it with the scheduler. Hence using timestamp makes your life easier.



Now we know the way to automate the sqoop job. Let’s discuss how to define a complete workflow for the sqoop incremental import. Below is the logic I used.


1. Create a separate folder path in the HDFS to get the data from RDBMS

Eg: /user/thilina_02338/source/ideabiz_admin/pin/charge/data

Each time when you import data, it’ll remove all files in this folder. It will prevent data duplication in the external tables. In my scenario, I don’t have replication database servers in the RDBMS. Hence I break up the data in the table wise. Then schedule it separately. This helps to minimize the load on the relational database. Hence I used separate path for each table. In the first time when you define the workflow, you need to do a lot of work. But this helps you to rectify the issue quickly.

For below queries, I’m using Hive UI. It’s capable of saving the query and import to the workflow as well. Also you can draw few graphs based on the results of the query.


2. Now create external table and point the folder path as similar to above. This will get all the data in the folder path to the external table.

CREATE EXTERNAL TABLE IF NOT EXISTS ext_pin_charge_txn (ID INT, DATETIME TIMESTAMP, APIMGR_ID INT, MSISDN varchar(20), APP_REF VARCHAR(100), SERVER_REF varchar(50), DESCRIPTION varchar(500), CALLBACK_URL varchar(500), AMOUNT double, TAX int, PIN varchar(7), RETRY_COUNT int, STATUS varchar(20), PAYMENT_STATUS varchar(20), PAYMENT_REF varchar(50), RESENT_COUNT int ) ROW FORMAT DELIMITED FIELDS TERMINATED BY '|' STORED AS TEXTFILE LOCATION '/user/thilina_02338/source/ideabiz_admin/pin/charge/data';


3. Create internal table which contains partition keys.

CREATE TABLE IF NOT EXISTS ideamart.pin_charge_txn (ID INT, DATETIME TIMESTAMP, APIMGR_ID INT, MSISDN varchar(20), APP_REF VARCHAR(100), SERVER_REF varchar(50), DESCRIPTION varchar(500), CALLBACK_URL varchar(500), AMOUNT double, TAX int, PIN varchar(7), RETRY_COUNT int, STATUS varchar(20), PAYMENT_STATUS varchar(20), PAYMENT_REF varchar(50), RESENT_COUNT int) PARTITIONED BY(time_key INT)


4. Insert data to the internal table from external table.

set hive.exec.dynamic.partition.mode=nonstrict;
insert into ideamart.pin_charge_txn partition(time_key)
select ID,DATETIME,APIMGR_ID,MSISDN,APP_REF,SERVER_REF,DESCRIPTION,CALLBACK_URL,AMOUNT,TAX,PIN,RETRY_COUNT,STATUS,PAYMENT_STATUS,PAYMENT_REF,RESENT_COUNT,cast(from_unixtime(unix_timestamp(cast(DATETIME as string),'yyyy-MM-dd HH:mm:ss'),'yyyyMM') as int)time_key
from ext_pin_charge_txn




The job scheduling can be done using coordinator application. Coordinator allows the user to execute recurrent and interdependent workflows.

That’s it! Yayyyyy. Scroll down and comment if you have any queries regarding incremental data import. Would love to help :)


Comments

Post a Comment

Popular posts from this blog

Hash Functions

-
Image
One way functions that transfer character inputs to compressed output value named as hash functions. The input can be infinite(arbitrary), but the output always has a finite(fixed) set of characters. This produces a fingerprint of the file/message/data source: Wikipedia You cannot reverse the function and get the input value in Hash functions. If you have the content, you can use the hash function to calculate the hash value, but the other way around is not possible. If you have the hash value and the function, you can’t use it to get the input text. Below are sample scenarios that hash functions may become useless. If the function is reversible, it’s not secure. Hash values can be exposed to the public. Hence we don’t want someone seeing the hash value. The values may not collision resistant. There may be use cases with two unique input values return the same hash value. Since digital certificates/Signatures using hash fun
Read more

How you can manage throttle out errors in WSO2 API Manager

-
Image
You might face the above-mentioned error if you have an application in Dialog Ideabiz platform.Let’s start from the beginning. Once you create an application on Ideabiz, you need to subscribe to the APIs you require. We use WSO2 API manager to manage the APIs . When admin approves your API subscriptions, he/she will assign a tier value for each API(5tps, 10 tps etc.). Also, they will assign a tier value for the application as well. Let’s assume you wanted to create an application by using SMS and Payment API. The team has assigned 5tps for each API and 10tps for your application. You use SMS API only to send SMS to the users and payment API to charge the users on daily basis. When considering each API, (as per above example) you should manage  5 transactions per second in the SMS API. Which means you can have 5 API calls per second. Let’s assume before you send the SMS, you want to charge Rs 1 from the user. Therefore you need to call payment API as well. Since the Pa
Read more

CAP THEOREM

-
Image
What’s C, A and P? Consistency is the easiest one. It roughly means that the clients get the same view of data. By saying that a system is consistent we often mean strong consistency, but it also can come in different flavors. Availability is a property saying that every request to a non-failing node will return a (meaningful) response. The response may not contain all data (so the harvest will not be 100%), but it should be useful for the client. Partition tolerance means that the system will continue working even if any number of messages sent between nodes is lost. This can be e.g. a network failure between two data centers, where nodes in each data center form a partition. Also note that a failure of any number of nodes forms a partition (it is not possible to distinguish between a network failure and a node failing and stopping to respond to messages). The basic statement of the CAP theorem is that, given the three properties of Consistency, Availability, and Pa
Read more