Does Argo support dynamic fan-out?

[justinmchase]

Is it possible for Argo to dynamically loop over a set which is generated by the output of a previous step?

So for example suppose I would like a workflow which processes a CSV file. For each line in that CSV file I would like to run a sub-workflow.

Or suppose I would like to process a video frame by frame, I want to fan-out every frame of the video and process each frame. I can't know how many frames there are, I need to essentially iterate the file to decode it and get the frames. Also there are far more frames then I can reasonably process in parallel so I need some way to limit the parallelism of the processing of the sub-frames. Once completed I would like to fan-in all of the results and re-encode a new video file.

I would really appreciate any examples of how to accomplish such a task.

[brews]

Sure!

Here is a simple example

# This workflow demonstrates the use of a generator step which produces a list of items as a result.
# This list is subsequently used for expanding the next step into multiple parallel steps.
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: loops-param-result-
spec:
  entrypoint: loop-param-result-example
  templates:
  - name: loop-param-result-example
    steps:
    - - name: generate
        template: gen-number-list
    - - name: sleep
        template: sleep-n-sec
        arguments:
          parameters:
          - name: seconds
            value: "{{item}}"
        withParam: "{{steps.generate.outputs.result}}"

  - name: gen-number-list
    script:
      image: python:alpine3.6
      command: [python]
      source: |
        import json
        import sys
        json.dump([i for i in range(20, 31)], sys.stdout)
  - name: sleep-n-sec
    inputs:
      parameters:
      - name: seconds
    container:
      image: alpine:latest
      command: [sh, -c]
      args: ["echo sleeping for {{inputs.parameters.seconds}} seconds; sleep {{inputs.parameters.seconds}}; echo done"]

from https://github.com/argoproj/argo-workflows/blob/master/examples/loops-param-result.yaml.

Looping and output-passing between steps or a dag are the key here. You could pass the data itself between steps/tasks or hand out references pointing to external data.

And here is a newer map-reduce example: https://github.com/argoproj/argo-workflows/blob/master/examples/map-reduce.yaml This has a fan-out and fan-in passing data around with an artifact repository.

[justinmchase]

Thank you this is perfect!

So in that last example they're dumping the id's json.dump(partIds, sys.stdout) which the subsequent jobs get fanned out to receive and then they use that to get to the artiafct, or theoretically a database record.

So how about parallelism, suppose I have in your example case here range(0, 1000000), will it try to create a million pods or is Argo smart enough to limit the parallel pods fanned out in this way?

Also, I believe argo visualizes the workflow run, would it show every single fanned out pod in this case? Or is the visualization smarter than that?

[hello-willy]

Sure!

Here is a simple example

# This workflow demonstrates the use of a generator step which produces a list of items as a result.
# This list is subsequently used for expanding the next step into multiple parallel steps.
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: loops-param-result-
spec:
  entrypoint: loop-param-result-example
  templates:
  - name: loop-param-result-example
    steps:
    - - name: generate
        template: gen-number-list
    - - name: sleep
        template: sleep-n-sec
        arguments:
          parameters:
          - name: seconds
            value: "{{item}}"
        withParam: "{{steps.generate.outputs.result}}"

  - name: gen-number-list
    script:
      image: python:alpine3.6
      command: [python]
      source: |
        import json
        import sys
        json.dump([i for i in range(20, 31)], sys.stdout)
  - name: sleep-n-sec
    inputs:
      parameters:
      - name: seconds
    container:
      image: alpine:latest
      command: [sh, -c]
      args: ["echo sleeping for {{inputs.parameters.seconds}} seconds; sleep {{inputs.parameters.seconds}}; echo done"]

from https://github.com/argoproj/argo-workflows/blob/master/examples/loops-param-result.yaml.

Looping and output-passing between steps or a dag are the key here. You could pass the data itself between steps/tasks or hand out references pointing to external data.

And here is a newer map-reduce example: https://github.com/argoproj/argo-workflows/blob/master/examples/map-reduce.yaml This has a fan-out and fan-in passing data around with an artifact repository.

hi argo expert:
i have a DAG diagram is shown below：

   A
   | 
B1    B2(loops result)
|       |
C1    C2(When B1 is completed, execute C1 immediately. There is no need to wait until both B1 and B2 are completed )

is argo supported? thank you for your help!

[alexec]

It v3 you can use data template too.

[justinmchase]

Can you point me to some documentation to understand what you're saying better?

[alexec]

https://github.com/argoproj/argo-workflows/blob/master/docs/data-sourcing-and-transformation.md