splitting messages based on content

[madalina2311]

Hi,
Is there a way to route messages to different folders based on a certain field inside the message?
I am reading from a nats stream and i want to put messages inside gcs . This part works fine but my messages have different payload and a further step is gcs->BigQ. My idea was to split gcs folder in subfolders and route messages to separate folders (route will happen based on a type field ). From there i can rely on BQ schema being generated on the fly (as all messages inside a folder will respect a certain envelope).
I am not sure if i can handle the splitting part in benthos or i should be looking at something else (like Nifi or Airflow).

[Jeffail]

Hey @madalina2311 yeah that should be possible with interpolation functions on the path field (https://www.benthos.dev/docs/components/outputs/gcp_cloud_storage#path):

You'd need to group the messages by the target folder before archiving them into the final file format, it'd look something like this:

output:
  gcp_cloud_storage:
    bucket: foo
    path: ${! meta("dir") }/${! uuid_v4() }.jsonl
    max_in_flight: 64
    batching:
      count: 1000
      processors:
        - group_by_value:
            value: ${! this.foo }
        - bloblang: 'meta dir = this.foo'
        - archive:
            format: lines

[madalina2311]

Thanks, I did the split easily with your hints. Really appreciate it !