basic.agg.yml

pipeline:
  batch_size: 1000

  source:
    type: kafka
    kafka:
      brokers: [localhost:9092]
      group_id: {{ SQLFLOW_GROUP_ID|default('test') }}
      auto_offset_reset: earliest
      topics:
        - {{ SQLFLOW_INPUT_TOPIC|default('topic-simple-agg') }}

  handler:
    type: 'handlers.InferredDiskBatch'
    sql: |
      SELECT
        properties.city as city,
        count(*) as city_count
      FROM batch
      GROUP BY
        city
      ORDER BY city DESC

  sink:
    type: kafka
    kafka:
      brokers: [localhost:9092]
      topic: {{ SQLFLOW_OUTPUT_TOPIC|default('output-simple-agg-disk') }}