ETL from LOCAL/REST to FTP/JSON.mql
ETL script with application 'supervision' compatibility
if (not (group exist "0001_folder1_folder2_folder3")) {
group add "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_1_source_connect.exe" true 5
(param
)
"Connect to the source"
{
# CONFIGURATION ;
-> "[CONF_NAME_OF_THE_FLOW]" "name_of_the_flow";
# INITIALIZATION ;
-> "[FLOW_PID]" [PID];
# HANDLE ;
try {
# Flow initialization;
stack flow_init [FLOW_PID] [CONF_NAME_OF_THE_FLOW] "{}";
stack flow_step [FLOW_PID] 1 "source_connect...";
# Local connection: don't use step 1;
log write "Source: Local connection" OK null null;
#Step 1 is valid;
stack flow_step [FLOW_PID] 1 "source_connect_ok";
include "folder1.folder2.folder3.step_2_extract.exe";
} {
#Step 1 is not valid;
stack flow_step [FLOW_PID] 1 "source_connect_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_1_source_connect.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_1_source_connect.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_2_extract.exe" true 1
(param
)
"Extract files from the source"
{
# HANDLE ;
try {
# Mark the flow as step 2;
stack flow_step [FLOW_PID] 2 "extract...";
# Create the local directory PID;
file mkdir (concat "home/" [FLOW_PID]);
#Step 2 is valid;
stack flow_step [FLOW_PID] 2 "extract_ok";
stack (date now) "folder1.folder2.folder3.step_3_transform.exe" "[FLOW_PID]" [FLOW_PID] "[filename]" "";
} {
#Step 2 is not valid;
stack flow_step [FLOW_PID] 2 "extract_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_2_extract.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_2_extract.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_3_transform.exe" true 1
(param
(var "[FLOW_PID]" {true} "The flow PID" is_null:true is_empty:true "1")
(var "[filename]" {true} "The filename" is_null:true is_empty:true "file.xxx")
)
"Transform the file"
{
# CONFIGURATION ;
-> "[CONF_DESTINATION_NEW_FILENAME]" (concat "new_filename_" (date systimestamp_min) ".json");
# HANDLE ;
try {
# Mark the flow as step 3;
stack flow_step [FLOW_PID] 3 "transform...";
json load "flow" "[]";
log write (concat "Build the file 'home/" [FLOW_PID] "/" [CONF_DESTINATION_NEW_FILENAME] "' from REST API...") OK null null;
json load "header" "{}";
json iobject "header" / "x-user" "admin" STR;
json iobject "header" / "x-password" "pwd" STR;
json load "flow_source" (rest https get "https://localhost:9999/api/addition" "v1=1&v2=6" (json doc "header") "[]");
# BEGIN METHOD 1;
-> "[T_A]" "";
-> "[T_B]" "";
-> "[T_C]" "";
json parse_obj "flow_source" "/" "[key]" "[val]" {
-> (concat "[T_" [key] "]") [val];
};
json load "row" "{}";
json iobject "row" / "A" [T_A] STR;
json iobject "row" / "B" [T_B] STR;
json iobject "row" / "C" [T_C] STR;
json iarray "flow" / (json doc "row") OBJ;
# END METHOD 1;
# BEGIN METHOD 2;
json parse_array "flow_source" "/" "row" {
-> "[T_A]" (json select "row" /A);
-> "[T_B]" (json select "row" /B);
-> "[T_C]" (json select "row" /C);
json load "row" "{}";
json iobject "row" / "A" [T_A] STR;
json iobject "row" / "B" [T_B] STR;
json iobject "row" / "C" [T_C] STR;
json iarray "flow" / (json doc "row") OBJ;
};
# END METHOD 2;
file create (concat "home/" [FLOW_PID] "/" [CONF_DESTINATION_NEW_FILENAME]) (json doc "flow");
log write (concat "Builded.") OK null null;
#Step 3 is valid;
stack flow_step [FLOW_PID] 3 "transform_ok";
stack (date now) "folder1.folder2.folder3.step_4_destination_connect.exe" "[FLOW_PID]" [FLOW_PID] "[filename]" [CONF_DESTINATION_NEW_FILENAME];
} {
#Step 3 is not valid;
stack flow_step [FLOW_PID] 3 "transform_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_3_transform.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_3_transform.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_4_destination_connect.exe" true 5
(param
(var "[FLOW_PID]" {true} "The flow PID" is_null:true is_empty:true "1")
(var "[filename]" {true} "The filename" is_null:true is_empty:true "file.xxx")
)
"Connect to the destination"
{
# CONFIGURATION ;
-> "[CONF_DESTINATION_CM]" "demo_cm";
# HANDLE ;
try {
# Flow initialization;
stack flow_step [FLOW_PID] 4 "destination_connect...";
ftp connect "destination_sess" {cm get [CONF_DESTINATION_CM];};
ftp active "destination_sess";
ftp passive "destination_sess";
ftp active compression "destination_sess" true;
ftp set type "destination_sess" "BINARY";
#Step 4 is valid;
stack flow_step [FLOW_PID] 4 "destination_connect_ok";
include "folder1.folder2.folder3.step_5_load.exe";
} {
try {ftp disconnect "destination_sess";} {} "[err]";
#Step 4 is not valid;
stack flow_step [FLOW_PID] 4 "destination_connect_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_4_destination_connect.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_4_destination_connect.exe" "0001_folder1_folder2_folder3";
};
script merge "folder1.folder2.folder3.step_5_load.exe" true 1
(param
)
"Load file to the destination"
{
# CONFIGURATION ;
-> "[CONF_DIR_DESTINATION]" "/remote/dir";
# HANDLE ;
try {
ftp cd "destination_sess" [CONF_DIR_DESTINATION];
# Mark the flow as step 5;
stack flow_step [FLOW_PID] 5 "load...";
log write (concat "Send '" (concat "home/" [FLOW_PID] "/" [filename]) "' to '" (concat [CONF_DIR_DESTINATION] "/" [filename]) "' ...") OK null null;
# Send file into the destination directory;
ftp put "destination_sess" (concat "home/" [FLOW_PID] "/" [filename]) "RESUME";
log write "Sended." OK null null;
ftp disconnect "destination_sess";
#Step 5 is valid;
stack flow_step [FLOW_PID] 5 "load_ok";
} {
try {ftp disconnect "destination_sess";} {} "[err]";
#Step 5 is not valid;
stack flow_step [FLOW_PID] 5 "load_KO";
# Give the error to the stack and stop the process;
exception (1) ([global_err]);
} "[global_err]";
} "Return nothing";
if (not (group is granted script "folder1.folder2.folder3.step_5_load.exe" "0001_folder1_folder2_folder3")) {
group grant script "folder1.folder2.folder3.step_5_load.exe" "0001_folder1_folder2_folder3";
};