ForEach activities iteration implementation in parallel?

Question

ForEach activities iteration implementation in parallel?

Andris Veidemanis 55

Hello,

Maybe this have changed, but I'm remembering that ADF used to iterate over array a bit differently then it does now.
Example pipeline:

{
    "name": "ForEach",
    "properties": {
        "activities": [
            {
                "name": "ForEach",
                "type": "ForEach",
                "dependsOn": [],
                "userProperties": [],
                "typeProperties": {
                    "items": {
                        "value": "@variables('v_foreach')",
                        "type": "Expression"
                    },
                    "isSequential": false,
                    "batchCount": 3,
                    "activities": [
                        {
                            "name": "Wait",
                            "type": "Wait",
                            "dependsOn": [],
                            "userProperties": [],
                            "typeProperties": {
                                "waitTimeInSeconds": {
                                    "value": "@item().WaitTime",
                                    "type": "Expression"
                                }
                            }
                        }
                    ]
                }
            }
        ],
        "variables": {
            "v_foreach": {
                "type": "Array",
                "defaultValue": [
                    {
                        "No": 1,
                        "WaitTime": 60
                    },
                    {
                        "No": 2,
                        "WaitTime": 1
                    },
                    {
                        "No": 3,
                        "WaitTime": 1
                    },
                    {
                        "No": 4,
                        "WaitTime": 60
                    },
                    {
                        "No": 5,
                        "WaitTime": 1
                    },
                    {
                        "No": 6,
                        "WaitTime": 1
                    },
                    {
                        "No": 7,
                        "WaitTime": 60
                    }
                ]
            }
        },
        "annotations": []
    }
}

In my mind this pipeline should run 63 seconds (removing ADF pipeline spin up time and such):

Time	Slot1	Slot2	Slot3
0 sec	"No": 1, "WaitTime: 60"	"No": 2, "WaitTime": 1	"No": 3, "WaitTime": 1
1 sec	"No": 1, "WaitTime": 59	"No": 4, "WaitTime": 60	"No": 5, "WaitTime": 1
2 sec	"No": 1, "WaitTime": 58	"No": 4, "WaitTime": 59	"No": 6, "WaitTime": 1
3 sec	"No": 1, "WaitTime": 57	"No": 4, "WaitTime": 58	"No": 7, "WaitTime": 60
...	...	...	...
62 sec	<empty>	<empty>	"No": 7, "WaitTime": 1
63 sec	<empty>	<empty>	<empty>

But in reality it is like this:

Time	Slot1	Slot2	Slot3
0 sec	"No": 1, "WaitTime": 60	"No": 2, "WaitTime": 1	"No": 3, "WaitTime": 1
1 sec	"No": 1, "WaitTime": 59	"No": 5, "WaitTime": 1	"No": 6, "WaitTime": 1
2 sec	"No": 1, "WaitTime": 58	<empty>	<empty>
...	...	...	...
60	"No": 4, "WaitTime": 60	<empty>	<empty>
...	...	...	...
120	"No": 7, "WaitTime": 60	<empty>	<empty>
...	...	...	...
179	"No": 7, "WaitTime": 1	<empty>	<empty>
180	<empty>	<empty>	<empty>

The time difference in this case is 117 seconds which in it self is nothing but in my case it is closer to two hours.

I understand that I can plan around it based on past times, but there is variations in activity times.

Is there a way to make ForEach work like in the first example? Has ForEach activity worked like this since the beginning?

Thanks!

QuantumCache 20,671 Reputation points Moderator

2023-07-12T18:27:24.9033333+00:00

Hello @Andris Veidemanis,

Thanks for posting a great discussion on For-Each activity!

How much time the For-Each has taken if you use Sequential mode?

The screenshots, are they from ADF pipeline run history? or something from customer tracking?

Parallel execution

If isSequential is set to false, the activity iterates in parallel with a maximum of 50 concurrent iterations. This setting should be used with caution. If the concurrent iterations are writing to the same folder but to different files, this approach is fine. If the concurrent iterations are writing concurrently to the exact same file, this approach most likely causes an error.

ADF: For-Each Type properties
Andris Veidemanis 55 Reputation points

2023-07-13T08:55:07.6833333+00:00
How much time the For-Each has taken if you use Sequential mode?

The example I'm using in Sequential mode would run 184 seconds.
That is the reason why I'm somewhat surprised about the way For-Each activity handles parallelization.
Because there is a subset of sub activity run times which totally beats the purpose of running activities in parallel.

In particular case I'm calling Custom activity and I wanted to For-Each activity to handle scheduling so to say and for each Custom activity running time be +/- the same as Task run time in Azure Batch job.

Any how - what I did:

Split For-Each inner activity to a separate pipeline

Set concurrency limit for that pipeline to 3

Set For-Each to run in batch mode with batchCount larger than total number of items to iterate over

and boom -> total run time 70 seconds

1 answer

Your answer

QuantumCache 20,671 Reputation points Moderator

2023-07-12T18:27:24.9033333+00:00

Hello @Andris Veidemanis,

Thanks for posting a great discussion on For-Each activity!

How much time the For-Each has taken if you use Sequential mode?

The screenshots, are they from ADF pipeline run history? or something from customer tracking?

Parallel execution

If isSequential is set to false, the activity iterates in parallel with a maximum of 50 concurrent iterations. This setting should be used with caution. If the concurrent iterations are writing to the same folder but to different files, this approach is fine. If the concurrent iterations are writing concurrently to the exact same file, this approach most likely causes an error.

ADF: For-Each Type properties
Andris Veidemanis 55 Reputation points

2023-07-13T08:55:07.6833333+00:00

How much time the For-Each has taken if you use Sequential mode?

The example I'm using in Sequential mode would run 184 seconds.
That is the reason why I'm somewhat surprised about the way For-Each activity handles parallelization.
Because there is a subset of sub activity run times which totally beats the purpose of running activities in parallel.

In particular case I'm calling Custom activity and I wanted to For-Each activity to handle scheduling so to say and for each Custom activity running time be +/- the same as Task run time in Azure Batch job.

Any how - what I did:

Split For-Each inner activity to a separate pipeline

Set concurrency limit for that pipeline to 3

Set For-Each to run in batch mode with batchCount larger than total number of items to iterate over

and boom -> total run time 70 seconds

Answer 1

Work around:

Split For-Each inner activity to a separate pipeline
Set concurrency limit for that pipeline to 3
Set For-Each to run in batch mode with batchCount larger than total number of items to iterate over

JSON for workaround:

{
    "name": "Wait",
    "properties": {
        "activities": [
            {
                "name": "Wait",
                "type": "Wait",
                "dependsOn": [],
                "userProperties": [],
                "typeProperties": {
                    "waitTimeInSeconds": {
                        "value": "@pipeline().parameters.p_WaitTime",
                        "type": "Expression"
                    }
                }
            }
        ],
        "concurrency": 3,
        "parameters": {
            "p_WaitTime": {
                "type": "int"
            }
        },
        "annotations": []
    }
}

{
    "name": "ForEach",
    "properties": {
        "activities": [
            {
                "name": "ForEach",
                "type": "ForEach",
                "dependsOn": [],
                "userProperties": [],
                "typeProperties": {
                    "items": {
                        "value": "@variables('v_foreach')",
                        "type": "Expression"
                    },
                    "isSequential": false,
                    "batchCount": 10,
                    "activities": [
                        {
                            "name": "Wait",
                            "type": "ExecutePipeline",
                            "dependsOn": [],
                            "userProperties": [],
                            "typeProperties": {
                                "pipeline": {
                                    "referenceName": "Wait",
                                    "type": "PipelineReference"
                                },
                                "waitOnCompletion": true,
                                "parameters": {
                                    "p_WaitTime": {
                                        "value": "@item().WaitTime",
                                        "type": "Expression"
                                    }
                                }
                            }
                        }
                    ]
                }
            }
        ],
        "variables": {
            "v_foreach": {
                "type": "Array",
                "defaultValue": [
                    {
                        "No": 1,
                        "WaitTime": 60
                    },
                    {
                        "No": 2,
                        "WaitTime": 1
                    },
                    {
                        "No": 3,
                        "WaitTime": 1
                    },
                    {
                        "No": 4,
                        "WaitTime": 60
                    },
                    {
                        "No": 5,
                        "WaitTime": 1
                    },
                    {
                        "No": 6,
                        "WaitTime": 1
                    },
                    {
                        "No": 7,
                        "WaitTime": 60
                    }
                ]
            }
        },
        "annotations": []
    }
}

QuantumCache 20,671 Reputation points Moderator

2023-07-13T15:52:22.31+00:00

Hello @Andris Veidemanis,

So the Parallel execution is now taking less time for execution? Looks like the issue is resolved!

Share via

ForEach activities iteration implementation in parallel?

1 answer

Your answer