Want to filter csv files based on condition in adf

Question

Want to filter csv files based on condition in adf

Amar Agnihotri 926

Hi,
I am having 300 csv files stored in adls container. Now i am trying to copy only 50 files out of 300 files to another container . Also out of those 50 files i want to filter some files based on some conditions like date >= '01-01-2019' for one file and Total count >= 100 for another file. Here, date is a column in one file and total count is a column in another file. I want to do this in adf. Earlier when i was having source data coming from Mysql db i did this successfully using lookup activity which was picking up file name and schema name from the look up file and i also defined the condition column in lookup file and then used If activity to apply multiple condition on different tables. But here the source is csv file and i don't know how to apply those condition on csv files. Please suggest something to get it done .

Subashri Vasudevan 11,231 Reputation points

2022-05-19T10:16:49.113+00:00
Hi @Amar Agnihotri ,

Just trying to understand where are you stuck?

extracting the condition from lookup file corresponding to the data file? For instance if the current file name is file1.txt, you need to extract the condition from the lookup file which may look like below

filename,condition
file1.txt,'lastmodifieddate>2022-01-31'
file2.txt,'rowcount>100'

Verifying the file content (rowcount or lastmodifieddt) based on the condition you get from the lookup file? (IF condition activity you mentioned)

Also, you say that your file brings the count and date columns. are these details part of every row, so that we just consider top 1 row for if condition ? Please provide sample file if possible

Please let us know your inputs for these to help you better.
Amar Agnihotri 926 Reputation points

2022-05-19T11:47:03.87+00:00

Hi @Subashri Vasudevan ,

Let me clear each step top you.
Take first scenario when source data is coming from Mysql db.

This is my lookup file stored in blob container

This helps ,me to pick only 4 tables mentioned in lookup file other wise the total number of tables in Mysqldb are 305. You can see that i have created a flag is_filtered and condition for each table in lookup file.

Now this is my pipeline -

The lookup activity is looking into the lookup file and just picking up only those tables then it is passing the output to for each activity . Inside foreach i am using if activity

You can see the expression of if activity when Is_filtered = 1 the true will execute and otherwise false will execute. Now if you will see the lookup file then If_filtered is having value 1 only for those tables on which i want to apply condition . Now this is inside true this is the copy activity where i wrote query in source to filter tables based on condition -

and this is the copy activity inside false where all the data of table will be copied
![203707-image.png][5]
Amar Agnihotri 926 Reputation points

2022-05-19T11:48:33.74+00:00

and this is the copy activity inside false where all the data of table will be copied

THIS PIPELINE IS WORKING FINE BUT NOW THE SOURCE DATA IS NOT MYSQL DB IT IS A CONTAINER HAVING 305 .CSV FILES ( SAME TABLES WHICH ARE IN MYSQL DB). NOW I WANT TO APPLY SAM,E LOGIC IN THE NEW PIPELINE . I AM STUCK AT THAT. WILL THE SAME LOGIC WORK FOR .CSV FILES ALSO ? HOPE YOU UNDERSTAND MY PROBLEM
Subashri Vasudevan 11,231 Reputation points

2022-05-19T14:48:35.487+00:00

Hi @Amar Agnihotri ,

In the copy activity you have used select statement to filter the source data.

In case of a csv source we don't have the flexibility to use select query. There is a workaround,

we can use lookup activity to read file contents and then a filter activity to filter the records.

If my source file is below -

sno,totalcount
1,100
2,100
3,200
4,700

and i want to filter the records where totalcount is 100, i can use the expression @equals(item().totalcount,'100')

this will filter the records and we will have the Value array with filtered records. (2 records in my case as follows)

{
"ItemsCount": 4,
"FilteredItemsCount": 2,
"Value": [
{
"sno": "1",
"totalcount": "100"
},
{
"sno": "2",
"totalcount": "100"
}
]
}

after that, we can choose to write the output as a json file in azure storage and consume that in the copy activity source.

Here is the reference to write filter activity output to a json file.

how-can-i-pass-output-from-a-filter-activity-directly-to-a-copy-activity-in-adf

Please let us know if this is of use.

5 answers

Your answer

Subashri Vasudevan 11,231 Reputation points

2022-05-19T10:16:49.113+00:00

Hi @Amar Agnihotri ,

Just trying to understand where are you stuck?

extracting the condition from lookup file corresponding to the data file? For instance if the current file name is file1.txt, you need to extract the condition from the lookup file which may look like below

filename,condition
file1.txt,'lastmodifieddate>2022-01-31'
file2.txt,'rowcount>100'

Verifying the file content (rowcount or lastmodifieddt) based on the condition you get from the lookup file? (IF condition activity you mentioned)

Also, you say that your file brings the count and date columns. are these details part of every row, so that we just consider top 1 row for if condition ? Please provide sample file if possible

Please let us know your inputs for these to help you better.
Amar Agnihotri 926 Reputation points

2022-05-19T11:47:03.87+00:00

Hi @Subashri Vasudevan ,

Let me clear each step top you.
Take first scenario when source data is coming from Mysql db.

This is my lookup file stored in blob container

This helps ,me to pick only 4 tables mentioned in lookup file other wise the total number of tables in Mysqldb are 305. You can see that i have created a flag is_filtered and condition for each table in lookup file.

Now this is my pipeline -

The lookup activity is looking into the lookup file and just picking up only those tables then it is passing the output to for each activity . Inside foreach i am using if activity

You can see the expression of if activity when Is_filtered = 1 the true will execute and otherwise false will execute. Now if you will see the lookup file then If_filtered is having value 1 only for those tables on which i want to apply condition . Now this is inside true this is the copy activity where i wrote query in source to filter tables based on condition -

and this is the copy activity inside false where all the data of table will be copied
![203707-image.png][5]
Amar Agnihotri 926 Reputation points

2022-05-19T11:48:33.74+00:00

and this is the copy activity inside false where all the data of table will be copied

THIS PIPELINE IS WORKING FINE BUT NOW THE SOURCE DATA IS NOT MYSQL DB IT IS A CONTAINER HAVING 305 .CSV FILES ( SAME TABLES WHICH ARE IN MYSQL DB). NOW I WANT TO APPLY SAM,E LOGIC IN THE NEW PIPELINE . I AM STUCK AT THAT. WILL THE SAME LOGIC WORK FOR .CSV FILES ALSO ? HOPE YOU UNDERSTAND MY PROBLEM
Subashri Vasudevan 11,231 Reputation points

2022-05-19T14:48:35.487+00:00

Hi @Amar Agnihotri ,

In the copy activity you have used select statement to filter the source data.

In case of a csv source we don't have the flexibility to use select query. There is a workaround,

we can use lookup activity to read file contents and then a filter activity to filter the records.

If my source file is below -

sno,totalcount
1,100
2,100
3,200
4,700

and i want to filter the records where totalcount is 100, i can use the expression @equals(item().totalcount,'100')

this will filter the records and we will have the Value array with filtered records. (2 records in my case as follows)

{
"ItemsCount": 4,
"FilteredItemsCount": 2,
"Value": [
{
"sno": "1",
"totalcount": "100"
},
{
"sno": "2",
"totalcount": "100"
}
]
}

after that, we can choose to write the output as a json file in azure storage and consume that in the copy activity source.

Here is the reference to write filter activity output to a json file.

how-can-i-pass-output-from-a-filter-activity-directly-to-a-copy-activity-in-adf

Please let us know if this is of use.

Answer 1

Nandan Hegde 36,156 MVP Volunteer Moderator

Hey,
Based on my understanding it is not possible directly via copy activity. You would need to use dataflow.

Below link might help:
https://learn.microsoft.com/en-us/answers/questions/552825/copy-activity-select-from-csv-dataset.html

Amar Agnihotri 926 Reputation points

2022-05-22T14:50:28.627+00:00

@Nandan Hegde

Hi,
As per your suggestion i tried to use data flow but still i am getting an error . I am showing my each step here -

This is my data source container containing 305 csv files -

out of these 305 files i want to copy only these 3 files to another container -

task.csv
user_location.csv
booking_by_day.csv

but while copying i want to filter these 3 csv files based on following conditions

condition for task.csv - date >= '2019-01-01'
condition for user_location - name= 'India'
condition for booking_by_day - booking_id>= 30000

now to select only three files out of 305 files and to apply the condition for different files i have created a lookup file having "Is_FILTERED" and CONDITION column as shown -

Now This is my pipeline

Unzip activity is just picking the 305 csv files from zip file and storing them to the container and then lookup activity is picking of the selected files using the loopkup file mentioned above -

Now i am using foreach to iterate through each file name coming from the lookup activity and inside for loop i am using if activity to apply the condition using dataflow. See here -

Answer 2

ShaikMaheer-MSFT 38,551 Microsoft Employee Moderator

Hi @Amar Agnihotri ,

Thank you for posting query in Microsoft Q&A Platform.

You can achieve this using below steps.
Step1: Lookup file on storage which tells us which files to process and along with conditions. use Lookup activity in ADF pipeline to read this file data.

Step2: Continue your foreach & If activities as it is.

Step3: In data flow activity have parameters to receive values from pipeline and inside Data flow activity pass values to those parameters.

Hope this helps. Please let us know if any further queries.

-----------

Please consider hitting Accept Answer. Accepted answers help community as well.

Amar Agnihotri 926 Reputation points

2022-05-23T11:25:55.513+00:00

Hi @ShaikMaheer-MSFT ,
First of all thanks for the reply . Now i have a couple of questions -
1)- As per ur suggestion i don't need to use filter activity in dataflow as i can mention condition in the parameter

2- my condition is different for different table as you can see in the snap of lookup file. What to do in that case. How to pass multiple condition for multiple table? Will parameter condition will grab condition from the lookup file automatically on when foreach iterator will grab the table name . I am Little confuse in this

3- Do i need to do anything in the sink in the dataflow to save the filtered table. In the sink filtered table will be copied alongwith all other tables on which conditions are not applied .

Answer 3

@Nandan Hegde

In the above snap you can see that If activity will check the condition IF_FILTERED =1 then true will execute otherwise FALSE will execute according to the column defined in lookup file as shown

and you can see that IS_FILTERED = 1 is only for task, user_location and booking_by_day csv so only the condition for these csv will be executed in the data flow to filter. This is the logic that i am trying to implement
Now inside the true activity i am calling the dataflow

and i want to pass the table name dynamically using the parameter defined in the dataflow -

In the red circle i want to pass the table name dynamically coming from the adf but don't know how to pass that . you suggested the blow link to take help from

https://learn.microsoft.com/en-us/answers/questions/552825/copy-activity-select-from-csv-dataset.html

but in that link foiltering is done on suing color and size condition but color and size condition is present in every table . In my scenario i want to filter three different tables using three different conditions for each table as defined in lookup file . Also the schema of all the tables are different . so i am stuck at these points -

1) - How to pass dynamic table names to dataflow on each execution of if activity for true condition ?
2) - and how to use filter condition between source and sink for three different tables .

Hope you understand my point now

Answer 4

@ShaikMaheer-MSFT ,
Hi there is one more thing to tell you. I tried too define parameters in dataflow as you did but don't know why i got this error.

Also want to tell you that for the data flow i am using dataset as a container which is having all the 305 csv files.

This is the source container for dataflow

It seems from your description that to define condition in a parameter in dataflow i have to select lookup file as source data set for dataflow but if i do so then what about the other files that i want to copy as it is in the sink . Please clear this

Answer 5

Amar Agnihotri 926

@ShaikMaheer-MSFT
Hi,

This is my adf pipeline and you can see that i have created parameter here

but i am not able to call these parameters inside dataflow

ShaikMaheer-MSFT 38,551 Reputation points Microsoft Employee Moderator

2022-05-25T06:40:30.043+00:00

Hi @Amar Agnihotri ,

Inside Dataflow activity from pipeline you are already passing values. Inside dataflow parameters you no need to add expressions again. We only need to pass values to these parameters from data flow activity thats it.

Please consider marking it as Accepted Answer. Accepted answers help community as well.
Amar Agnihotri 926 Reputation points

2022-05-26T11:59:21.057+00:00

@ShaikMaheer-MSFT hi,

i tried again only with the given parameters but i think my dataflow is not communicating with the source dataset. see the attached gif
ShaikMaheer-MSFT 38,551 Reputation points Microsoft Employee Moderator

2022-05-27T05:19:28.827+00:00

Hi @@Amar Agnihotri ,

You are passing values to parameters. But I don't see you are using that parameters inside data flow any where. Is your parameters going to point source file to pick then in that case under source options in the wild card path options you should use your parameters as per requirement.

From the error messages it seems the file or path to which you are referring inside source transformation --> source options --> wild card path is not exists or incorrect. Kindly check it once.

Share via

Want to filter csv files based on condition in adf

5 answers

Your answer