Trigger processor based on the condition - Nifi

[#1]

I want to create a generic single flow which should run for multiple tables and will upload the result flow files to google cloud and azure blob .

I have created the flow setup on nifi but what I want to achieve is that I dont want the azure upload for all the tables .Azure upload should execute only for specific tables and upload to google cloud for all the tables .

Any idea on how to achieve the conditional trigger based on any flag ?

Thanks

[#2 · Accepted solution]

@aman mittal

You can use RouteOnAttribute Processor to route on the required tables into AzureBlobStorage ..etc.

Below is the sample flow for demonstration

In this flow we are doing listing all the tables and Generating sql queries that fetches pages then ExecutingSql queries that are generated by GenerateTableFetch Processor.

As you need to load all the tables to GoogleCloud so fork the Success relation from ExecuteSql, in this step we are storing all the tables into Google Cloud.

In another fork from ExecuteSql processor feed the Success relationship to RouteOnAttribute Processor and add new property as

AzureBlob Tables

${db.table.name:equals("tablename1","tablename2")} //db.table.name attribute is added from List database table processor 

(or)

${db.table.name:toLower():equals("tablename1","tablename2")} //table names are not case sensitive in this EL

Use your attribute that having table name in the place of (db.table.name attribute) and Add all the required tables for azure in the above list.

Then use the same property name relationship to feed to Azure Blob Storage processor.

-

If the Answer addressed your question, Click on Accept button below to accept the answer, That would be great help to Community users to find solution quickly for these kind of issues.

[#3]

Thanks alot for such a great help 1 it works !!

[#4]

@Shu is there any way we can have run time forking based on any of the flag . Because I need to do this type of flow for more than 2k tables in my db.

[#5]

@Shu_ashu I tried this below expression for two job names in my flow and it is throwing error. Could you help?

Options I tried:

```${N_JOB:equalsIgnoreCase('DWDMBL0D','DWDMBL0F'):and(${LAST_RUN:equals('${custom_date}')}):and(${C_CURR_STA:equalsIgnoreCase('C')})}```

```${N_JOB:equalsIgnoreCase("DWDMBL0D","DWDMBL0F"):and(${LAST_RUN:equals('${custom_date}')}):and(${C_CURR_STA:equalsIgnoreCase('C')})}```