hive sqoop increment

create database ccdm_mstr;


use ccdm_mstr;


create table clm_cvs_fact(

clm_cvs_fact_ket varchar(30),

dw_cret_aud_key int(25),

dw_updt_aud_key int(30));


insert into clm_cvs_fact (clm_cvs_fact_ket, dw_cret_aud_key, dw_updt_aud_key) values ('123456700','12347771', '12347771');

insert into clm_cvs_fact (clm_cvs_fact_ket, dw_cret_aud_key, dw_updt_aud_key) values ('123456701','12347772', '12347772');


 select * from clm_cvs_fact;

 

 output:

+------------------+-----------------+-----------------+

| clm_cvs_fact_ket | dw_cret_aud_key | dw_updt_aud_key |

+------------------+-----------------+-----------------+

| 123456700        |        12347771 | 12347771        |

| 123456701        |        12347772 | 12347772        |

+------------------+-----------------+-----------------+



[cloudera@quickstart ~]$ sqoop import --connect jdbc:mysql://localhost/ccdm_mstr --username root --password cloudera --table clm_cvs_fact -m 1 --target-dir /user/ccdm_mstr_test --incremental append --check-column dw_updt_aud_key --last-value 0


[cloudera@quickstart ~]$ hadoop fs -cat /user/ccdm_mstr_test/part-m-00000

123456700,12347771,12347771

123456701,12347772,12347772


Now creted a table in hive 

[cloudera@quickstart ~]$ hive

create if not exist database ccdm_mstr;

use ccdm_mstr;

create table clm_cvs_fact

(

clm_cvs_fact_ket String,

dw_cret_aud_key String,

dw_updt_aud_key String

)

row format delimited

fields terminated by ','

location '/user/ccdm_mstr_test';


output:

select * from ccdm_mstr.clm_cvs_fact;

123456700 12347771 12347771

123456701 12347772 12347772

Time taken: 0.654 seconds, Fetched: 2 row(s)

==================

Trying to insert third record 


insert into clm_cvs_fact (clm_cvs_fact_ket, dw_cret_aud_key, dw_updt_aud_key) values ('123456702','12347773', '12347773');


mysql> select * from clm_cvs_fact;+------------------+-----------------+-----------------+

| clm_cvs_fact_ket | dw_cret_aud_key | dw_updt_aud_key |

+------------------+-----------------+-----------------+

| 123456700        |        12347771 | 12347771        |

| 123456701        |        12347772 | 12347772        |

| 123456702        |        12347773 | 12347773        |

+------------------+-----------------+-----------------+

=============

Now we are will runing sqoop with incremental load

sqoop import --connect jdbc:mysql://localhost/ccdm_mstr --username root --password cloudera --table clm_cvs_fact -m 1 --target-dir /user/ccdm_mstr_test --incremental append --check-column dw_updt_aud_key --last-value "12347772"


hive > select * from ccdm_mstr.clm_cvs_fact;

OK

123456700 12347771 12347771

123456701 12347772 12347772

123456702 12347773 12347773


================

Trying to update,delete,insert


insert -->

insert into clm_cvs_fact (clm_cvs_fact_ket, dw_cret_aud_key, dw_updt_aud_key) values ('123456705','12347776', '12347776');


update-->12342777 -->12349999


UPDATE clm_cvs_fact  SET dw_updt_aud_key = '12349999' WHERE clm_cvs_fact_ket = 123456701;


mysql> select * from clm_cvs_fact;+------------------+-----------------+-----------------+

| clm_cvs_fact_ket | dw_cret_aud_key | dw_updt_aud_key |

+------------------+-----------------+-----------------+

| 123456700        |        12347771 | 12347771       |

| 123456701        |        12347772 | 12399999       |

| 123456702        |        12347773 | 12347773        |

| 123456705        |        12347776 | 12347776        |

+------------------+-----------------+-----------------+


==============

Again running sqoop to import all new record 


sqoop import --connect jdbc:mysql://localhost/ccdm_mstr --username root --password cloudera --table clm_cvs_fact -m 1 --target-dir /user/ccdm_mstr_test --incremental append --check-column dw_updt_aud_key --last-value "12347773"


hive> 

    > 

    > select * from ccdm_mstr.clm_cvs_fact;

OK

123456700 12347771 12347771

123456701 12347772 12347772

123456702 12347773 12347773

123456701 12347772 12349999

123456705 12347776 12346777

Time taken: 0.056 seconds, Fetched: 5 row(s)


select t1.* from (select *  from clm_cvs_fact) t1 join (select clm_cvs_fact_ket, max(dw_updt_aud_key) md from  (select  * from clm_cvs_fact) t2 group by clm_cvs_fact_ket) s ON t1.clm_cvs_fact_ket = s.clm_cvs_fact_ket AND t1.dw_updt_aud_key = s.md;


hive output:


123456700 12347771 12347771

123456701 12347772 12349999

123456702 12347773 12347773

123456705 12347776 12347776


Comments

Post a Comment

Popular posts from this blog

Shell encrypt and decrypt

   #!/bin/sh       # path to virtualenv cd /home/srvc_nextgen_hitg/cp360 source cp360venv/bin/activate echo "Virtualenv started" cd /home/cp360/python_scripts #Read python code export AES_SECRET_KEY="K{;5%A5yHL&^efe-" export apiacc="yen.why.saw-99" encrypt_pswd=`python - <<END import os from AES import AES_ENCRYPT aes_obj = AES_ENCRYPT() encpt = aes_obj.encrypt(os.environ['apiacc'],os.environ['AES_SECRET_KEY']) print(encpt) END`  echo "------encrypt_pswd------------" echo $encrypt_pswd echo "-------------------------------" export encrypt_pswd=$encrypt_pswd decrypt_pswd=`python - <<END import os from AES import AES_ENCRYPT aes_obj = AES_ENCRYPT() dencpt = aes_obj.decrypt(os.environ['encrypt_pswd'],os.environ['AES_SECRET_KEY']) print(dencpt) END`  echo "------decrypt_pswd------------" echo $decrypt_pswd echo "-------------------------------" #############################...
Read more

cdc

 select  CONCAT(TRIM(ASSET.asset_number),0),asset.data_sou) AS cdc_key ,CONCAT(nvl(TRIM(asset.asset_number),'0'),asset.data_sou) AS dat_sou ,md5(concat(nvl(asset.asset_number,'0'),nvl(asset.dat_sou,'0'),nvl(asset.coln,'0') AS cdc_hash ,asset.* from ( select distinct CONCAT(trim(cus_nu),'-'), cast(NULL as string) as vendor_id, ACUR.*, 'N' ISDELETED from INERMIDTE.ACUR )ASSET ===================================== WITH ADDTION AS ( SELECT A.*  ,current_timestamp AS Date_created1, ,current_timestamp AS Date_updated1, ,'N' AS ISDELTED1 FROM precdc.asset A left outer join processed.asset B ON (A.cdc_key = B.cdc_key)  WHERE B.data_sou='1002' AND  (B.cdc_hash IS Null) ) ,DELETION  AS ( SELECT B.* ,B.current_timestamp AS Date_created1, ,B.current_timestamp AS Date_updated1, ,'Y' AS ISDELTED1 FROM precdc.asset A rigt outer join processed.asset B ON (A.cdc_key = B.cdc_key)  WHERE B.data_sou='1002' AND  (B.cdc_hash IS ...
Read more

incrementally update

Incrementally update an imported table In  CDP Private Cloud Base , updating imported tables involves importing incremental changes made to the original table using Sqoop and then merging changes with the tables imported into Hive. After ingesting data from an operational database to Hive, you usually need to set up a process for periodically synchronizing the imported table with the operational database table. The base table is a Hive-managed table that was created during the first data ingestion. Incrementally updating Hive tables from operational database systems involves merging the base table and change records to reflect the latest record set. You create the incremental table as a Hive external table, typically from CSV data in HDFS, to store the change records. This external table contains the changes (INSERTs and UPDATEs) from the operational database since the last data ingestion. Generally, the table is partitioned and only the latest partition is updated, making this pro...
Read more