Step 4: Configure Batching

Batching collects a number of messages and sends them as a group, reducing the number of output operations.

The Goal

You will modify your pipeline's output to use a batching policy. Instead of writing every single log to a file as it arrives, the pipeline will collect 10 messages (or wait for 5 seconds) and then write them all at once.

The batching Block

This example places the batching policy on a broker wrapping the file output. The two most common fields are:

• count: The number of messages to collect before sending.
• period: The maximum amount of time to wait before sending, even if the count has not been reached.

Whichever condition is met first triggers the send.

Implementation

1. Start with the Previous Pipeline: Copy the restructure-log.yaml from Step 3 to a new file named batched-output.yaml.

   cp restructure-log.yaml batched-output.yaml

2. Add the Batching Logic: Open batched-output.yaml and replace the entire output section with the batching broker and its file output below.

   Partial snippet

   Replace the 'output' section in batched-output.yaml

   output:
     broker:
       pattern: fan_out
       batching:
         count: 10
         period: 5s
       outputs:
         - file:
             path: /tmp/batched_logs.jsonl
             codec: lines

3. Deploy and Test:

   # Create the output file's directory if it doesn't exist
   mkdir -p /tmp

4. Verify: Watch the output file. With one generated message per second, the 5-second period flushes the batch before it reaches 10 messages.

   tail -f /tmp/batched_logs.jsonl

   If you were to stop the input, any remaining messages in the batch would be flushed after the 5-second period expires.

You have now implemented a batching policy, which is a critical technique for building high-throughput, cost-effective data pipelines.