ETL from MENTDB/CSV to LOCAL/SQL.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";
-> "[CONF_SOURCE_CM]" "demo_cm";
# 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...";
tunnel connect "session_remote" {cm get [CONF_SOURCE_CM];};
#Step 1 is valid;
stack flow_step [FLOW_PID] 1 "source_connect_ok";
include "folder1.folder2.folder3.step_2_extract.exe";
} {
try {tunnel disconnect "session_remote";} {} "[err]";
#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"
{
# CONFIGURATION ;
-> "[CONF_DIR_SOURCE]" "tmp/source";
-> "[CONF_SOURCE_FILTER]" ".*csv";
# 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]);
# Get files into PID directory;
log write (concat "Get files from '" [CONF_DIR_SOURCE] "/" [CONF_SOURCE_FILTER] "'.") OK null null;
json load "valid_files" (tunnel execute "session_remote" (concat
"-> \"[CONF_DIR_SOURCE]\" \"" (mql encode [CONF_DIR_SOURCE]) "\";"
"-> \"[CONF_SOURCE_FILTER]\" \"" (mql encode [CONF_SOURCE_FILTER]) "\";"
(mql {
file dir_list_regex [CONF_DIR_SOURCE] [CONF_SOURCE_FILTER] true false;
})
));
json parse_array "valid_files" "/" "[filename]" {
tunnel execute "session_remote" (concat
"-> \"[CONF_DIR_SOURCE]\" \"" (mql encode [CONF_DIR_SOURCE]) "\";"
"-> \"[filename]\" \"" (mql encode [filename]) "\";"
(mql {
file reader_open "r1" (concat [CONF_DIR_SOURCE] "/" [filename]) BINARY null;
})
);
file writer_open "w1" (concat "home/" [FLOW_PID] "/" [filename]) true BINARY null;
#Parse the file;
while (is not null (-> "[bytes]" (tunnel execute "session_remote" "file reader_get_bytes \"r1\" 402400"))) {
file writer_add_bytes "w1" [bytes];
file writer_flush "w1";
};
#Close the reader and the writer;
tunnel execute "session_remote" "file reader_close \"r1\"";
file writer_close "w1";
log write (concat "Get file 'home/" [FLOW_PID] "/" [filename] "'.") OK null null;
};
# Delete source file;
json parse_array "valid_files" "/" "[filename]" {
tunnel execute "session_remote" (concat
"-> \"[CONF_DIR_SOURCE]\" \"" (mql encode [CONF_DIR_SOURCE]) "\";"
"-> \"[filename]\" \"" (mql encode [filename]) "\";"
(mql {
file delete (concat [CONF_DIR_SOURCE] "/" [filename]);
})
);
log write (concat "Remove source file '" [CONF_DIR_SOURCE] "/" [filename] "'.") OK null null;
};
tunnel disconnect "session_remote";
} {
try {tunnel disconnect "session_remote";} {} "[err]";
#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]";
log write (concat (json count "valid_files" /) " file(s) to transform.") OK null null;
if (> (json count "valid_files" /) 0) {
#Step 2 is valid;
stack flow_step [FLOW_PID] 2 (concat "extract_ok_" (json count "valid_files" /) "_files");
json parse_array "valid_files" "/" "obj" {
-> "[filename]" (json select "obj" /name);
stack (date now) "folder1.folder2.folder3.step_3_transform.exe" "[FLOW_PID]" [FLOW_PID] "[filename]" [filename];
};
} {
#Step 2 is valid;
stack flow_step [FLOW_PID] 2 "extract_ok_zero_file";
# No file into the directory;
file delete (concat "home/" [FLOW_PID]);
};
} "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.csv")
)
"Transform the file"
{
# HANDLE ;
try {
# Mark the flow as step 3;
stack flow_step [FLOW_PID] 3 "transform...";
# SQL: don't use step 3;
log write "SQL: don't use step 3" 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]" [filename];
} {
#Step 5 is not valid;
stack flow_step [FLOW_PID] 5 "transform_and_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_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_SQL]" "MENTDB";
# HANDLE ;
try {
# Flow initialization;
stack flow_step [FLOW_PID] 4 "destination_connect...";
sql connect "destination_sess" {cm get [CONF_DESTINATION_CM_SQL];};
sql auto_commit "destination_sess" false;
#Step 4 is valid;
stack flow_step [FLOW_PID] 4 "destination_connect_ok";
include "folder1.folder2.folder3.step_5_load.exe";
} {
try {sql 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_SOURCE_CSV_SEPARATOR]" ",";
-> "[CONF_SOURCE_CSV_QUOTE]" "'";
-> "[CONF_SOURCE_CSV_FORCE_COLUMN]" "A,B,C";
-> "[CONF_DESTINATION_CM_SQL]" "MENTDB";
-> "[CONF_DESTINATION_COMMIT_BY_NB_ROW]" 5000;
# HANDLE ;
try {
# Mark the flow as step 5;
stack flow_step [FLOW_PID] 5 "load...";
log write (concat "Integrate the file 'home/" [FLOW_PID] "/" [filename] "' into the database '" [CONF_DESTINATION_CM_SQL] "'...") OK null null;
-> "[index]" 0;
-> "[index_all]" 0;
csv parse (mql "T") (concat "home/" [FLOW_PID] "/" [filename]) (mql [CONF_SOURCE_CSV_SEPARATOR]) (mql [CONF_SOURCE_CSV_QUOTE]) (mql [CONF_SOURCE_CSV_FORCE_COLUMN]) {
sql dml "destination_sess" (concat
"INSERT INTO `test` (
`A`,
`B`,
`C`
) VALUES (
" (sql encode [T_A]) " ,
" (sql encode [T_B]) " ,
" (sql encode [T_C]) "
);"
);
++ "[index]";
++ "[index_all]";
if (== [index] [CONF_DESTINATION_COMMIT_BY_NB_ROW]) {
-> "[index]" 0;
sql commit "destination_sess";
log write (concat "Current row >>> " [index_all]) OK null null;
};
};
sql commit "destination_sess";
sql disconnect "destination_sess";
log write (concat "Integrated.") OK null null;
#Step 5 is valid;
stack flow_step [FLOW_PID] 5 "load_ok";
} {
try {sql rollback "destination_sess";} {} "[err]";
try {sql 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";
};