Dispatch a pod for each entry in CSV

[Seanny123]

I use Argo Workflows to dispatch lists of jobs defined in a CSV. I accomplish this by chaining a bunch templates together, which involves:

• Breaking up the CSV file into individual JSON objects
• Parsing the JSON into parameters
• Actually passing the parameters to individual pods

I didn't see an example of this anywhere, so I thought I would share it here. The YAML which accomplishes this is:

entrypoint: main

templates:
  - name: main
    steps:
      - - name: get-inputs
          # Produces the complete set of work units from the initial input
          template: split-csv
          arguments:
            artifacts:
              - name: csv-file
                s3: retrieve csv from s3

      - - name: process-each
          # Iterates over the a set of work units produced by the previous step
          template: compute-one
          arguments:
            parameters:
              - name: index
                value: "{{item}}"
            artifacts:
              - name: mappings
                from: "{{steps.get-inputs.outputs.artifacts.json-data}}"
          withSequence:
            count: "{{steps.get-inputs.outputs.parameters.length}}"

  - name: compute-one
    # Processes a single work unit
    inputs:
      parameters:
        - name: index
      artifacts:
        - name: mappings
    steps:
      - - name: get-work-item
          # Retrieves the artifact references that are required to process a single unit of work
          template: get-work-item
          arguments:
            parameters:
              - name: index
                value: "{{inputs.parameters.index}}"
            artifacts:
              - name: mappings
                from: "{{inputs.artifacts.mappings}}"

      - - name: big-compute
          # Where the parameters accessed from the CSV get used.
          template: my-compute-job
          arguments:
            parameters:
              - name: param0
                value: "{{steps.get-work-item.outputs.parameters.param0}}"
              - name: param1
                value: "{{steps.get-work-item.outputs.parameters.param1}}"
              - name: param2
                value: "{{steps.get-work-item.outputs.parameters.param2}}"

  - name: get-work-item
    # From a given JSON array, get the item at `index`, which is expected to be an object,
    # and output the values of its keys to pass as parameters
    inputs:
      parameters:
        - name: index
      artifacts:
        - name: mappings
          path: /tmp/mappings.json
    outputs:
      parameters:
        - name: param0
          valueFrom:
            path: /tmp/param0
        - name: param1
          valueFrom:
            path: /tmp/param1
        - name: param2
          valueFrom:
            path: /tmp/param2
    script:
      image: stedolan/jq
      command: [sh]
      source: |
        jq -r '.[{{inputs.parameters.index}}].param0' {{inputs.artifacts.mappings.path}} > {{outputs.parameters.param0.path}}
        jq -r '.[{{inputs.parameters.index}}].param1' {{inputs.artifacts.mappings.path}} > {{outputs.parameters.param1.path}}
        jq -r '.[{{inputs.parameters.index}}].param2' {{inputs.artifacts.mappings.path}} > {{outputs.parameters.param2.path}}

  - name: split-csv
    # Given a CSV file, convert each row into a JSON-formatted object,
    # and output the list all resulting objects as an artifact,
    # and the length of this list as a parameter
    inputs:
      artifacts:
        - name: csv-file
          path: /tmp/input.csv
    script:
      image: python:alpine
      command: [python]
      source: |
        from csv import reader
        import json

        with open("{{inputs.artifacts.csv-file.path}}", "r") as f:
          rows = reader(f)
          next(rows)
          data = [ {"param0": r[0], "param1": r[1], "param2": r[2]} for r in list(rows) ]

        with open("{{outputs.artifacts.json-data.path}}", "w") as f:
          f.write(json.dumps(data))

        with open("{{outputs.parameters.length.path}}", "w") as f:
          f.write(str(len(data)))
    outputs:
      parameters:
        - name: length
          valueFrom:
            path: /tmp/length
      artifacts:
        - name: json-data
          path: /tmp/data.json

If you have a better way of implementing it, feel free to submit an answer to my question on CodeReview StackExchange.