how to convert message to batch message for parquet_encode in output

[fearfate]

example config like this:

pipeline:
  processors:
  - mapping: |
      root = [{"key1": "value1"}, {"key1": "value2"}, {"key": "value2"}] 
  - unarchive:
      format: json_array
output:
  switch:
    cases:
    - check: |
        this.exists("key1")
      output:
        broker:
          copies: 1
          pattern: fan_out
          outputs:
          - file:
              path: example-${! timestamp_unix_nano() }.parquet
              codec: all-bytes
         batching:
            byte_size: 104857600
            period: 3m
            processors:
            - parquet_encode:
                schema:
                - name: key1
                  type: UTF8
    - check: |
         this.exists("key")
       output:
         drop: {}

but I will got two parquet file, and I tried group_by_value and archive: format: json_array, not work as i was expected. Is there any way to convert message to batch message in output?

[mihaitodor]

Hey @fearfate, I'm not sure what's going wrong. Can you add a log processor after parquet_encode with message: ${! error() } to see if the processor might fail? For example, STRING isn't a valid value for the type in the parquet_encode schema.

[fearfate]

it's true the mapping has some error, and the batch processors parquet_encode encode failed, which leader to single message to file.