Below are the list of Talend jobs which were developed as part of IRM .

Talend Project :  DATA_OCEAN_DOMAIN_PROCUREMENT

GCP Project : prj-data-dm-procurement-prod

Schedule : All this jobs are scheduled to run weekly (yet the moment its configured to run on Monday's , but can be changed based on the need)

Job Explanation: Available at the bottom of the page.

We are using the Data Ocean standard jobs to extract and load the data into the STG/ODS tables. As we are dealing with Full and incremental type of loading so there is a little variance in the jobs, which is explained below:

Full Load

For the explanation, I am using F001_SPF_F001_F_W_ESLL_TO_BQ which is a full load job.  Job parameters like STG table, ODS table , Bucket details , filters etc are maintained in the RDS parameters and its not hardcoded in the job, if required it can be changed any time in the RDS itself.

Parameters Snapshot

• All parameters are self explanatory except the additional_filter which is used to define the type of the load i.e. blank means its a full refresh and "incremental" mean its a incremental load type.  See the below example for PF1_MSEG which is a incremental load type, so we used "incremental" in the additional_filter parameter.

Job Details

Below is the snapshot of the main Flow job, where we are using all the Data Ocean standard jobs , except "SAP to GCS" extraction based on each source.

J020_SPF_Table_ESLL_to_GCS  Job which is used to extract the data from the SAP table and load into the Google Cloud Storage, as I said all the parameters are configured in the RDS and no hardcoding of any values in the Talend job.

***************************************************************************************************************************************************************************************************************************************************************************************

Incremental Load

For the explanation, I am using F001_SPF_F001_I_W_MSEG_TO_BQ which is a incremental load job.  Job parameters like STG table, ODS table , Bucket details , filters etc are maintained in the RDS parameters and its not hardcoded in the job, if required it can be changed any time in the RDS itself.

Parameters Snapshot

• All parameters are self explanatory except the additional_filter which is used to define the type of the load i.e. blank means its a full refresh and "incremental" mean its a incremental load type.  See the below example for PF1_MSEG which is a incremental load type, so we used "incremental" in the additional_filter parameter.

Job Details

Below is the snapshot of the main Flow job, where we are using all the Data Ocean standard jobs , except "SAP to GCS" extraction based on each source, 

J020_SPF_Table_MSEG_to_GCS  Job which is used to extract the data from the SAP table and load into the Google Cloud Storage, as I said all the parameters are configured in the RDS and no hardcoding of any values in the Talend job.

The only difference between the Full load and incremental load is incremental load will have additional sub jobs to deal with last execution datatime, this can be used to fetch the delta records.  These jobs are standard jobs and the parameters values are sent from the main flow job

• T101_GET_LAST_EXTRACT : This job takes the l_VAR_PF1_MSEG_INC_LOAD parameter value and query to in the prj-data-dm-procurement-prod.STG.incremental_loading table to fetch the value in the meta_last_process_date column, which stores the last executed datetime from the delta column. And this value is being used in the tJava_1 where we construct the query which is passed to the extraction job in the below parameters

• T102_UPDATE_LAST_EXTRACT : This job is used to get the max("meta_stg_insert_date") from the STG table and update the prj-data-dm-procurement-prod.STG.incremental_loading table for that SOURCE_TABLE name. If you look at the below snapshot, where the value in meta_last_process_date tells the date & time till when we have extracted the records from the source table. In the next run, the above job will use this value and construct the filter condition so the job will extract the records which are inserted after this datetime.