Trigger any Google Service based on Google Cloud Storage Events

Recently, I encountered a requirement that involved triggering a Google Cloud Composer DAG based on events occurring in Google Cloud Storage (GCS).

To illustrate this scenario, let's consider an example: Suppose I have a Composer DAG designed to process data files stored in a GCS bucket. These files may arrive at any time throughout the day, and it's crucial to process them promptly upon arrival since they contain time-sensitive information.

Now, let's explore several solutions to address this requirement:

Option 1: Scheduled DAG Runs

Scheduling a DAG to run at regular intervals throughout the day is a straightforward approach. However, it suffers from several drawbacks:

1. Potential Overkill: Running the DAG every few minutes, regardless of whether new files are available, can lead to unnecessary resource consumption and costs, especially if the frequency far exceeds the actual arrival rate of files.

2. Processing Delay: If a file arrives just after a DAG run completes, it will have to wait until the next scheduled run, potentially introducing processing delays of up to the interval period.

For scenarios where files arrive sporadically or with varying frequencies, Option 1 may not be the most efficient solution.

Option 2: Cloud Function with GCS Trigger

Creating a Cloud Function triggered by GCS events offers a more responsive and efficient solution:

1. Real-Time Processing: By triggering the Cloud Function upon file creation or finalization in GCS, we ensure immediate processing of incoming files, eliminating unnecessary delays.

2. Dynamic DAG Submission: The Cloud Function can dynamically determine the appropriate DAG to submit based on the incoming file's metadata, allowing for flexibility and automation.

However, this approach has a notable drawback:

1. Potential Overhead: While the Cloud Function provides real-time processing capabilities, it triggers for every file that arrives in the designated bucket, potentially leading to unnecessary function invocations and associated costs.

Option 3: Pub/Sub Notifications for GCS

Utilizing Pub/Sub notifications for GCS object changes introduces a more refined and scalable solution:

1. Fine-Grained Triggering: By subscribing a Cloud Function to a Pub/Sub topic associated with specific GCS events (e.g., object finalization in a designated folder), we can ensure that the function only triggers when relevant files are created or modified.

2. Efficient Resource Utilization: With Pub/Sub notifications, we avoid the overhead of continuously polling GCS for changes and running unnecessary function invocations, leading to more efficient resource utilization and cost savings.

3. Real-Time Processing with Reduced Overhead: This approach combines the benefits of real-time file processing with reduced overhead, making it an optimal choice for scenarios where responsiveness and efficiency are paramount.

i). Create a topic on the GCS bucket, folder( ex: gs://gc-us-gcs-xyz-bkt/stg/) Ref: https://cloud.google.com/storage/docs/gsutil/commands/notification Sample Code: gsutil notification create -t tpc_xyz_file_stg -f json -e OBJECT_FINALIZE

-p stg/ gs://gc-us-gcs-xyg-bkt ii). Deploy the Cloud Function and associate the Cloud Function to the topic created above. Refer Cloud Function creation documentation to create a Cloud Function. Sample Code: gcloud functions deploy xyz_trigger_cf_gcs_events --runtime python312

--region us-east4

--ingress-settings=internal-and-gclb --entry-point xyz_trigger_cf_on_gcs_events

--trigger-topic= tpc_xyz_file_stg --source .

Conclusion:

While each option has its merits, Option 3 stands out as the most appropriate solution for triggering Composer DAGs based on GCS events. By leveraging Pub/Sub notifications, we achieve real-time processing with minimal overhead and optimal resource utilization, ensuring timely and efficient handling of time-sensitive data files.

REST API to trigger Airflow/Composer DAG

REST APIs are a popular way for applications to communicate with each other, allowing for flexible and scalable interactions. In this blog post, we'll explore how to use Google Cloud Run to trigger an Airflow/Composer DAG through a REST API.

Google Cloud Run is a fully managed compute platform that allows you to run stateless containers that are automatically scaled to meet incoming traffic. Airflow is an open-source platform to programmatically author, schedule and monitor workflows. Composer is a managed version of Airflow by Google Cloud.

Before diving into the implementation, let's first understand what a DAG is. A DAG (Directed Acyclic Graph) is a collection of tasks that are dependent on each other, with a defined order of execution. Airflow uses DAGs to define workflows and their dependencies.

To trigger an Airflow/Composer DAG using a REST API, we can create a simple Python Flask app that sends a request to the Airflow/Composer REST API endpoint. The REST API endpoint can be used to trigger a DAG run by passing a JSON payload that contains the DAG ID, and any other necessary parameters.

Here's an example of how to create a Flask app to trigger a DAG using the Airflow REST API:

 from flask import Flask, request

import requests

app = Flask(__name__)

# Airflow/Composer REST API endpoint

AIRFLOW_ENDPOINT = 'https://airflow.googleapis.com/v1beta1/dags/<dag_id>/dagRuns'

# Service account credentials

HEADERS = {

    'Authorization': 'Bearer <access_token>',

    'Content-Type': 'application/json'

}

@app.route('/trigger-dag', methods=['POST'])

def trigger_dag():

    # Get payload data from request body

    payload = request.get_json()

    # Build request data

    data = {

        'conf': payload.get('conf', {}),

        'replace_microseconds': 'false'

    }

    # Send request to Airflow REST API

    response = requests.post(AIRFLOW_ENDPOINT, json=data, headers=HEADERS)

    if response.status_code == 200:

        return 'DAG triggered successfully', 200

    else:

        return 'Failed to trigger DAG', response.status_code

In this example, we have defined a route '/trigger-dag' that accepts a POST request with a JSON payload. The payload contains any necessary parameters for the DAG run. The Flask app sends a request to the Airflow/Composer REST API endpoint using the requests library.

The HEADERS variable contains the service account credentials required to authenticate with the Airflow/Composer REST API. Replace <access_token> with the actual access token for the service account that has the necessary permissions to trigger DAGs.

Replace <dag_id> with the actual ID of the DAG that you want to trigger. You can find the DAG ID in the Airflow/Composer UI or by running the command airflow list_dags in the Cloud Shell.

Once you have created the Flask app, you can deploy it to Google Cloud Run using the following command:

gcloud run deploy <service_name> --image gcr.io/<project_id>/<image_name> --platform managed

Replace <service_name> with the desired name for the Cloud Run service, <project_id> with your Google Cloud project ID, and <image_name> with the name of the container image that contains the Flask app.

Finally, you can test the REST API by sending a POST request to the Cloud Run service URL with the JSON payload. The DAG should trigger successfully, and you can monitor the status of the DAG run in the Airflow/Composer UI.

In conclusion, using Google Cloud Run to trigger an Airflow/Composer DAG through a REST API is a simple and scalable solution

You may check this code repository for working solution

Working code

PL/SQL REST API Call to receive JSON Payload

 declare

  req utl_http.req;

  res utl_http.resp;

  lv_url varchar2(4000) := 'http://<host>:<port>/<something>/';

  buffer varchar2(32767);

  lj_json_obj  json_object_t;

BEGIN

dbms_output.put_line(lv_url);

req := utl_http.begin_request(lv_url,'POST','HTTP/1.1');

                 utl_http.set_header(req,'Content-type','application/json');

                 utl_http.set_header(req, 'Accept', 'application/json');

                 res := utl_http.get_response(req);

  -- process the response from the HTTP call

  begin

      utl_http.read_text(res, buffer,1000000);

      lj_json_obj := json_object_t(buffer);

      utl_http.end_response(res);

  exception

    when utl_http.end_of_body then

      utl_http.end_response(res);

     end ;

end if;

end;

/

Python- Pandas

Read individual field values from a data frame

db_df = pd.read_sql("select brms_file_check_notify_pkg.decrypt(a.login_id) password_dr, 

a.* from file_check_notification a"                    " where sysdate > next_run and file_check_id = 3"                       , ebsconn)
ebsconn.close()
print('DB is queried for tab1..')
hostname = db_df.loc[0].at['host']

username = db_df.loc[0].at['login_id']
password = db_df.loc[0].at['password_dr']
location = db_df.loc[0].at['location']
subject  = db_df.loc[0].at['email_subject']
protocol = db_df.loc[0].at['protocol_type']

Numbers to Words Converter

Oracle Analytic Functions - Difference between ORDER BY and PARTITION BY

ORDER BY - brings up the running total/sum by certain repeating column
PARTITION BY - brings up the GROUP total by certain repeating column

Consider following simple table.

This projects example contains 2 projects - ABC and CAB
Each project contains 2 tasks and its cost respectively.

By using same Analytic functions with two different window, can get "running totals" and "group totals"

select project_id, project_name, task_no, cost
,sum(cost) over(order by project_id,task_no) running_total_by_project_task
,sum(cost) over(order by project_id) running_total_by_project
,sum(cost) over(partition by project_id) total_cost_by_project
,sum(cost) over(partition by task_no) total_cost_by_task
from temp_project_details;

Output below..

Getting Concurrent Request Id within Sql Loader based Concurrent Programs

None of the following will work in SQL LOADER.
- fnd_global.conc_request_id
- fnd_profile.value('conc_request_id')

First will update request_id as -1 and second will update as blank/NULL

You may consider this.. it works for me.

- Create a function, something like below...
===
CREATE OR REPLACE PACKAGE BODY

IS

-- define global variable..

   gn_sqlldr_req_id     NUMBER ;

FUNCTION get_request_id RETURN NUMBER
IS
ln_req_id NUMBER ;
BEGIN


IF gn_sqlldr_req_id IS NULL then

select max(request_id)
INTO gn_sqlldr_req_id
from fnd_concurrent_requests where  concurrent_program_id  in (
select concurrent_program_id from fnd_concurrent_programs where concurrent_program_name =' YOUR_PROG_SHORT_NAME')   ;
END IF ;

RETURN gn_sqlldr_req_id ;

END ;

====
Your SQL Loader control file..
==
load data
INFILE '/tmp/XMITINVD.TXT'
INTO TABLE xxinv_TABLE_stg
APPEND
FIELDS TERMINATED BY '|'
TRAILING NULLCOLS

(

---

---

---

request_id ".get_request_id" -- your function here

--

--

--

)

This worked for me. 

How to divide huge data into batches

Here is my answer on the subject above.

Using an Analytical function – NTILE, I was able to divide big volume of data into multiple batches.

After dividing data into multiple batches, it’s easy to implement multi-threading, without missing any record.

Following query divides Items(for the Org: 489 ) data in to 5 equal batches, and returns MIN and MAX Item ids from each batch.

 select min(inventory_item_id) min_item_id , max(inventory_item_id) max_item_id,

     count(*) batch_count, batch batch_num

    from ( select inventory_item_id, NTILE(5) over (order by inventory_item_id) batch

               from mtl_system_items_b

               where organization_id  = 679 -- 048

            )    group by batch

Query Output:

MIN_ITEM_ID	MAX_ITEM_ID	BATCH_COUNT	BATCH_NUM
158281	299596	7092	1
299598	400344	7092	2
509671	589229	7091	4
589230	961093	7091	5
400345	509659	7092	3

I used this batching concept to process my data into multi-threading. Sample code below.

Sample Code:

---

for rec in (select min(inventory_item_id) min_item_id , max(inventory_item_id) max_item_id,

count(*) batch_count, batch batch_num

from ( select inventory_item_id, ntile(5) over (order by inventory_item_id) batch

                      from mtl_system_items_b

                      where organization_id  = <489>)         group by batch

) loop

               print.log_message ('batch_num:'||rec.batch_num

                              ||':min_item_id:' ||rec.min_item_id

                              ||':max_item_id:'||rec.max_item_id

                              ||':batch_count:'||rec.batch_count);

               ln_request_id := 0;

begin

               -- submit programs in batchhes..

      ln_request_id :=

               fnd_request.submit_request (

                   application   => 'XXCUST',

                   program       => 'concp_short_name',

                   argument1     => xv_item_code,

                   argument2     => rec.min_item_id,

                   argument3     => rec.max_item_id,            

                   argument4     => xn_org_id          

                   );

              

commit;

end loop ;

--

BI Publisher Bursting - Delivering/sending emails conditionally

Here is the sample code to send emails conditionally from BI/XML Publisher bursting feature.

Problem Statement:

Generate payslips for all the employees at one shot and email the payslips to respective employee.

1. Alter EMP table and added a column called email to hold emails of each individual employees.

And then update email column to add email Ids where you want your emails to be delivered.

2. Create a Data template Sample code available here

3. Default package Sample code available here

4.Define a concurrent program and assign the concurrent program to a responsibility as per your convenience.

5. Design a simple RTF template and register the same. Sample template available here

6. Register data template, created in #1 above

7. Register control file for bursting. Sample code available here

Package Sample code here.

8. Submit the concurrent program defined in #4. If everything goes fine,
you should be able to see emails delivered at the email ids provided.

Sample Data in XML format.

9. Sample output below..

Extensible Atrributes

Here is the query to list out all extensible attribute group names, attribute names and its data types.

SELECT   *
    FROM (SELECT --egoattributeeo.attr_id,
                 egoattributeeo.application_id,
                 egoattributeeo.attr_group_type,
                 egoattributeeo.attr_group_name,
                  egoattributeeo.attr_name,
                 egoattributeeo.attr_display_name,
                 decode(egoattributeeo.data_type_code,'C','CHAR','N','NUM','X','DATE') data_type_code,
                 egoattributeeo.DEFAULT_VALUE,
                 egoattributeeo.value_set_name,
                 egoattributeeo.maximum_size,
                  egoattributeeo.enabled_flag,
                  egoattributeeo.required_flag,
                 egoattributeeo.database_column,
                 egoattributeeo.read_only_flag
            FROM ego_attrs_v egoattributeeo, ego_fnd_dsc_flx_ctx_ext ext
           WHERE egoattributeeo.application_id = ext.application_id
             AND egoattributeeo.attr_group_type =
                                                ext.descriptive_flexfield_name
             AND egoattributeeo.attr_group_name =
                                             ext.descriptive_flex_context_code) qrslt
   WHERE (application_id = AND attr_group_type like '%'
        )
ORDER BY attr_group_type, attr_group_name



-- All the attribute groups --
SELECT FL_CTX_EXT.ATTR_GROUP_ID ATTR_GROUP_ID ,
FL_CTX.APPLICATION_ID APPLICATION_ID ,
FL_CTX.DESCRIPTIVE_FLEXFIELD_NAME ATTR_GROUP_TYPE ,
FL_CTX.DESCRIPTIVE_FLEX_CONTEXT_CODE ATTR_GROUP_NAME ,
TL.DESCRIPTIVE_FLEX_CONTEXT_NAME ATTR_GROUP_DISP_NAME ,
TL.DESCRIPTION DESCRIPTION ,
FL_CTX.ENABLED_FLAG ENABLED_CODE ,
L1.MEANING ENABLED_MEANING ,
FL_CTX_EXT.MULTI_ROW MULTI_ROW_CODE ,L2.MEANING MULTI_ROW_MEANING ,
FL_CTX_EXT.VIEW_PRIVILEGE_ID VIEW_PRIVILEGE ,
FUNC_VIEW_TL.USER_FUNCTION_NAME VIEW_PRIVILEGE_NAME ,
FL_CTX_EXT.EDIT_PRIVILEGE_ID EDIT_PRIVILEGE ,
FUNC_EDIT_TL.USER_FUNCTION_NAME EDIT_PRIVILEGE_NAME ,
FL_CTX_EXT.AGV_NAME AGV_NAME ,
FL_CTX_EXT.REGION_CODE REGION_CODE ,
FL_CTX_EXT.BUSINESS_EVENT_FLAG BUSINESS_EVENT_FLAG ,
L3.MEANING BUSINESS_EVENT_MEANING ,
'N' IS_EDITABLE ,
FL_TL.TITLE AGT_DISP_NAME ,
FL_CTX_EXT.PRE_BUSINESS_EVENT_FLAG PRE_BUSINESS_EVENT_FLAG ,
L4.MEANING PRE_BUSINESS_EVENT_MEANING ,'N' IS_DELETEABLE
FROM FND_DESCR_FLEX_CONTEXTS FL_CTX ,
EGO_FND_DSC_FLX_CTX_EXT FL_CTX_EXT ,
FND_DESCR_FLEX_CONTEXTS_TL TL ,
FND_DESCRIPTIVE_FLEXS_TL FL_TL ,
FND_LOOKUP_VALUES L1 ,
FND_LOOKUP_VALUES L2 ,
FND_LOOKUP_VALUES L3 ,
FND_LOOKUP_VALUES L4 ,
FND_FORM_FUNCTIONS_TL FUNC_VIEW_TL ,
FND_FORM_FUNCTIONS_TL FUNC_EDIT_TL
WHERE FL_CTX.APPLICATION_ID = FL_CTX_EXT.APPLICATION_ID AND
FL_CTX.APPLICATION_ID = TL.APPLICATION_ID AND
FL_CTX.DESCRIPTIVE_FLEXFIELD_NAME = FL_CTX_EXT.DESCRIPTIVE_FLEXFIELD_NAME AND
FL_CTX.DESCRIPTIVE_FLEXFIELD_NAME = TL.DESCRIPTIVE_FLEXFIELD_NAME AND
FL_CTX.DESCRIPTIVE_FLEXFIELD_NAME = FL_TL.DESCRIPTIVE_FLEXFIELD_NAME AND
FL_CTX.DESCRIPTIVE_FLEX_CONTEXT_CODE = FL_CTX_EXT.DESCRIPTIVE_FLEX_CONTEXT_CODE AND
FL_CTX.DESCRIPTIVE_FLEX_CONTEXT_CODE = TL.DESCRIPTIVE_FLEX_CONTEXT_CODE AND
TL.LANGUAGE = USERENV('LANG') AND FL_TL.LANGUAGE = USERENV('LANG') AND
L1.LOOKUP_TYPE = 'YES_NO' AND
L1.LOOKUP_CODE = FL_CTX.ENABLED_FLAG AND
L1.LANGUAGE = USERENV('LANG') AND
L1.VIEW_APPLICATION_ID = 0 AND
L2.LOOKUP_TYPE = 'YES_NO' AND
L2.LOOKUP_CODE = FL_CTX_EXT.MULTI_ROW AND
L2.LANGUAGE = USERENV('LANG') AND
L2.VIEW_APPLICATION_ID = 0 AND
L3.LOOKUP_TYPE(+) = 'YES_NO' AND
L3.LOOKUP_CODE (+)= FL_CTX_EXT.BUSINESS_EVENT_FLAG AND
L3.LANGUAGE (+)= USERENV('LANG') AND
L3.VIEW_APPLICATION_ID(+) = 0 AND
L4.LOOKUP_TYPE(+) = 'YES_NO' AND
L4.LOOKUP_CODE (+)= FL_CTX_EXT.PRE_BUSINESS_EVENT_FLAG AND
L4.LANGUAGE (+)= USERENV('LANG') AND
L4.VIEW_APPLICATION_ID(+) = 0 AND
FL_CTX_EXT.VIEW_PRIVILEGE_ID = FUNC_VIEW_TL.FUNCTION_ID(+) AND
FUNC_VIEW_TL.LANGUAGE (+)= userenv('LANG') AND
FL_CTX_EXT.EDIT_PRIVILEGE_ID = FUNC_EDIT_TL.FUNCTION_ID(+) AND
FUNC_EDIT_TL.LANGUAGE (+)= userenv('LANG') AND
FL_CTX_EXT.DESCRIPTIVE_FLEX_CONTEXT_CODE NOT IN ('ItemDetailImage', 'ItemDetailDesc')
--and FL_CTX.DESCRIPTIVE_FLEX_CONTEXT_CODE = 'PIM_CLF'

When updating Supplier Address, oracle.apps.ap.supplier.event was not raised

Problem: Business event "oracle.apps.ap.supplier.event" may not be raised when you updating Supplier Address at site level.

Solution:  You need to check the checkbox 'Update to all sites using this address'.

This business event "oracle.apps.ap.supplier.event" will be raised when you do following actions on suppliers/vendors
- Create Vendor
- Update Vendor
- Create Vendor Site
- Update Vendor Site
- Create Vendor Contact
- Update Vendor Contact
- Update Address Assignments

Critical Tasks in Project Management - typically ERP Implementation

Hi All,

I have put this list together from my past and previous experience and exposure as Project Lead and Manager. Many of these may or may not be applicable depending Project type, Rollout or upgrade, Implemetation. But ideally these points mentioned below are usually missed and or not under consideration until the situation arises. But project management practise says that one should always be prepared for the worst even before it shows up at the door step. I have seen on several occasions when we as a team had to go back and review the strategy to mitigate the risk of missing the timeline

Project plan should include the below activities

1 Check for prerequisite version of 11i needed for R12 upgrade

2 DB upgrade for supporting/as needed to 10g or 11g

3 Server capacity ( RAM, HDD) increase as needed to support 10g or 11g accordingly

4 Necessary pre-requisite patches to be applied for R12 as given for the corresponding version and research on the new patches released for that version and modules related.

5 Data Conversion/Migration planning

6 Instance planning and availability plan

7 Requirement gathering

8 Requirement signoff

9 Module Related Patches Verification

10 Cut Over Activities

11 Seeded Functional Setup steps.

12 Custom components functional design documents

13 Data Conversion/Migration Functional Design

14 Customization Technical design documents

15 Data Conversion/Migration Technical Design

16 Code Build

17 Unit Testing

18 System Integration Testing

19 User acceptance testing

20 Regression Testing

21 Volume Testing

22 Pre Production activities

23 Cut Over Activities

To top it all these should have bookmarks in the Risk Management document. Ideally each step can be flagged off red depending on the stage or the current status of the project affairs.

Finding Guarantors Case Folder for a Parent case folder

Below is the query to find Guarantors/Child case folder for a Parent/Dealer case folder along credit analyst assigned to it.



select child_cf.case_folder_number child_case_folder, parent_cf.case_folder_number  parent_case_folder,  res.source_name
from
ar_cmgt_credit_requests re,
ar_cmgt_case_folders child_cf,
ar_cmgt_case_folders parent_cf,
jtf_rs_resource_extns res
where
parent_credit_request_id = parent_cf.credit_request_id
--and parent_cf.case_folder_number = '14225'
and res.source_name = 'Credit Analyst Name Here'
and parent_cf.review_type is not null
and res.resource_id = parent_cf.credit_analyst_id
and child_cf.credit_request_id = re.credit_request_id
and child_cf.review_type is not null
order by 2 desc

Finding Credit Analyst Name in Oracle Credit Management

-- To find Credit analyst name for the given case folder name. ------

select source_name credit_analyst_ name, source_job_title Job_Title, case_folder_number, usr.user_name User, usr.description
from jtf_rs_resource_extns res ,
ar_cmgt_case_folders cf,
fnd_user usr
where
res.resource_id = cf.credit_analyst_id
and case_folder_number = '' 
and res.user_id = usr.user_id


  -- ====== to find out credit analyst  assigned to Rule Name -----

select rule_name, result_value , usr.user_name credit_analyst_name , res.resource_id
from fun_rule_details  fun, jtf_rs_resource_extns res, fnd_user usr
where rule_name = '' and
fun.result_value = res.resource_id
and res.user_id = usr.user_id