How to run parallel pipelines

[eduardo-fp-romao]

Hello o/

I'm having a little problem how to configure the pipeline with the parallel. I have 4 custom processors: "employee_mapper", "employee_reporter", "info_mapper", "info_reporter". The processer "employee_reporter "needs to run after the "employee_mapper" (we pass the result to report) and the same for "info" but the "employee group" can run in parallel with the info. I tried something like this:

pipeline:
  threads: 4
  processors:
    - parallel:
        pipeline:
          processors:
            - type: employee_mapper
            - type: employee_reporter
        pipeline:
          processors:
            - type: info_mapper
            - type: info_reporter

Unfortunately I get this error:

Configuration file read error: ..\config\benthos.yaml: (18,1) yaml: unmarshal errors:
  line 23: mapping key "pipeline" already defined at line 19

How can I configure the pipeline?

[mihaitodor]

Hey @eduardo-fp-romao 👋 I think you're confusing a few things here. The pipeline is a top level config section which lets you run the sequence of processors under pipeline.processors in parallel against batches of messages emitted by your input(s) as dictated by pipeline.threads. If you don't configure any batching, then individual messages are considered batches of size 1. You can, additionally, use a parallel processor in your pipeline.processors which can have one or more child processors under the processors field and it also has a cap field which lets you control how many individual messages from a batch it should process in parallel.

In your case, if the employee_reporter processor needs to run after the employee_mapper processor, then it's fine to have them one after the other in pipeline.processors. The goal of parallelism in Benthos is to process messages in parallel rather than run multiple processors in parallel for a given message. There are ways to achieve that if needed, but I didn't need to do anything like that so far.

[eduardo-fp-romao]

Oh... I think I understand, so basically for each batch of message I can execute multiple task. For me I have 1 message that will be mapped to multiple objects, so I can do something like this:

pipeline:
  threads: 4
  processors:
    - type: get_message (example,)
    - parallel:
          processors:
            - type: employee_mapper
            - type: info_mapper
    - type: employee_reporter
    - type: info_reporter

or even

pipeline:
  threads: 4
  processors:
    - type: get_message (example)
    - parallel:
          processors:
            - type: employee_mapper
            - type: info_mapper
    - parallel:
          processors:
            - type: employee_reporter
            - type: info_reporter

And than use the cache resource to pass the rights arguments to the each reports