kafka-json-mutating.yaml

input:
  type: broker
  broker:
    count: 8 # Try cranking this value up if your CPUs aren't maxed out
    inputs:
    - type: kafka_balanced
      kafka_balanced:
        addresses:
        - localhost:9092 # TODO
        client_id: benthos_mutator_1
        consumer_group: benthos_mutator_group
        topics:
        - data_stream
  processors:
  - type: combine
    combine:
      parts: 8 # Batch size: Tune this to increase output throughput
pipeline:
  threads: 4 # This should exactly match the number of logical CPUs
  processors:
  - type: jmespath
    jmespath:
      query: |
        {
          content: content,
          entities: sentiment[?confidence > `0.3`].{
            name: entity,
            sentiment: level
          },
          language: max_by(language, &confidence).code
        }
output:
  type: kafka
  kafka:
    addresses:
    - localhost:9092 # TODO
    client_id: benthos_mutator_1
    topic: mutated_data_stream