entechlog

    ansible_user: ec2-user							   |	    ansible_user: nadesansiva
    ansible_ssh_private_key_file: /tmp/certs/ssh_priv.pem			   |	    #ansible_ssh_private_key_file: /tmp/certs/ssh_priv.pem
										   >	    #   no_proxy: "{{ lookup('env','env_no_proxy') }}"
										   >
										   >	    ### All custom paths
										   >	    ksql_jmx_configmap_path: /opt/prometheus/
										   >	    ksql_udf_path: /opt/confluent/ext/
										   >	    ksql_javatmp_path: /opt/confluent/javatmp/
										   >	    ksql_state_path: /opt/confluent/ksqlstate/
										   >	    ksql_rocksdbtmp_path: /opt/confluent/rocksdbtmp/

entechlog

-- Insert one more record in source stream and inspect the result
INSERT INTO STM_DEPARTMENTS_RAW (DEPT_NO, DEPT_NAME) VALUES ('d002', 'Claims');

SELECT * FROM STM_DEPARTMENTS_TGT EMIT CHANGES;
+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+
|ROWTIME                   |ROWKEY                    |HUMAN_TS                  |DEPT_NO                   |CURR_DEPT_NAME            |PREV_DEPT_NAME            |
+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+
|1587515912933             |d001                      |2020-04-21 20:38:32 -0400 |d001                      |Marketing                 |null                      |
|1587515912955             |d002                      |2020-04-21 20:38:32 -0400 |d002                      |Claims                    |null                      |
|15875159

entechlog

-- Insert one more record in source stream and inspect the result
INSERT INTO STM_DEPARTMENTS_RAW (DEPT_NO, DEPT_NAME) VALUES ('d001', 'Marketing Team');

ksql> select * from STM_DEPARTMENTS_TGT emit changes;
+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+
|ROWTIME                   |ROWKEY                    |HUMAN_TS                  |DEPT_NO                   |CURR_DEPT_NAME            |PREV_DEPT_NAME            |
+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+--------------------------+
|1587515912933             |d001                      |2020-04-21 20:38:32 -0400 |d001                      |Marketing                 |null                      |
|1587515912955             |d002                      |2020-04-21 20:38:32 -0400 |d002                      |Claims                    |null                   

entechlog

TBL_DEMO_DROP_KSQL_OBJECT_0030,yes
STM_DEMO_DROP_KSQL_OBJECT_0010,no
STM_DEMO_DROP_KSQL_OBJECT_0020,yes
STM_DEMO_DROP_KSQL_OBJECT_0020,yes

entechlog

2020-04-1523:19:02 - dropKqlDBObjects Started
 
**************************************************************************
2020-04-1523:19:02 - Processing :  TBL_DEMO_DROP_KSQL_OBJECT_0030
**************************************************************************

Issuing Command (TERM) :  curl -s -X "POST" localhost:8088/ksql -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" -d '{"ksql": "DESCRIBE TBL_DEMO_DROP_KSQL_OBJECT_0030;"}' \ | grep -zoP '"id":\s*\K[^\s,]*(?=\s*,)' | sed 's/"//g' | sed 's/\x0//g' | xargs -Iq1 curl -s -X "POST" localhost:8088/ksql -H "Content-Type: application/vnd.ksql.v1+json; charset=utf-8" -d '{"ksql": "TERMINATE 'q1';"}'
[{"@type":"currentStatus","statementText":"TERMINATE CTAS_TBL_DEMO_DROP_KSQL_OBJECT_0030_636;","commandId":"terminate/CTAS_TBL_DEMO_DROP_KSQL_OBJECT_0030_636/execute","commandStatus":{"status":"SUCCESS","message":"Query terminated."},"commandSequenceNumber":637,"warnings":[]}]
INFO                   : QUERIES FOR TBL_DEMO_DROP_KSQL_OBJECT_0030 TERMINA

entechlog

**************************************************************************
SCRIPT_NAME            : dropKqlDBObjects
**************************************************************************
INFO                   : logs directory exists
INFO                   : temp directory exists
INFO                   : reports directory exists
INFO                   : Total arguments.. 3
INFO                   : Received the correct number of input parameters
INFO-MODE              : sed
INFO-KSQL_URL          : localhost:8088

entechlog

CREATE STREAM STM_DEMO_DROP_KSQL_OBJECT_0010
	WITH (KAFKA_TOPIC = 'dev.etl.datagen.orders.src.0010', VALUE_FORMAT = 'Avro');

CREATE STREAM STM_DEMO_DROP_KSQL_OBJECT_0020 AS

SELECT CUSTOMERID, AS_MAP(AS_ARRAY('orderID', 'StorePickUpFlag', 'ItemName', 'OrderDate'), AS_ARRAY(ORDERID, STOREPICKUPFLAG, ITEMNAME, ORDERDATE)) AS orderData
FROM STM_DEMO_DROP_KSQL_OBJECT_0010 PARTITION BY CUSTOMERID;

CREATE TABLE TBL_DEMO_DROP_KSQL_OBJECT_0030 AS

entechlog

SELECT TIMESTAMPTOSTRING(CURR.ROWTIME, 'yyyy-MM-dd HH:mm:ss Z'), CURR.KSQL_COL_1, PREV.KSQL_COL_1
FROM TBL_ARRAY_DEMO_0030 CURR
INNER JOIN TBL_ARRAY_DEMO_0010 PREV ON CURR.ROWKEY = PREV.ROWKEY
WHERE TRIM(UCASE(CURR.CUSTOMERID)) = '6454214343400541'
	AND CURR.KSQL_COL_1 <> PREV.KSQL_COL_1 EMIT CHANGES;

entechlog

INSERT INTO TBL_ARRAY_DEMO_0010 (CUSTOMERID, KSQL_COL_1)
VALUES ('6454214343400541', AS_ARRAY('{orderID=799997, ItemName=Arlo Pro, StorePickUpFlag=Yes, OrderDate=06/24/2019}', '{orderID=799992, ItemName=Apple TV 4K Streaming Console, StorePickUpFlag=No, OrderDate=01/28/2020}'));

INSERT INTO TBL_ARRAY_DEMO_0030 (CUSTOMERID, KSQL_COL_1)
VALUES ('6454214343400541', AS_ARRAY('{orderID=799997, ItemName=Arlo Pro, StorePickUpFlag=Yes, OrderDate=06/24/2019}', '{orderID=799992, ItemName=Apple TV 4K Streaming Console, StorePickUpFlag=No, OrderDate=01/28/2020}'));

entechlog

CREATE TABLE TBL_ARRAY_DEMO_0030 (CUSTOMERID VARCHAR(STRING), KSQL_COL_1 ARRAY < VARCHAR(STRING) >)
	WITH (KAFKA_TOPIC = 'TBL_ARRAY_DEMO_0030', VALUE_FORMAT = 'AVRO', KEY = 'CUSTOMERID', PARTITIONS = 1, REPLICAS = 1);