kestra_duckdb_event_based.yml

id: salesReport
namespace: dev
tasks:
  - id: downloadCsv
    type: io.kestra.plugin.fs.http.Download
    uri: https://gist.githubusercontent.com/anna-geller/15f19626d975877b40c3653b6745dcd6/raw/849e8f69a251ece8bfb32dbd6097e69af6fa7f7f/orders.csv

  - id: analyzeSales
    type: io.kestra.plugin.jdbc.duckdb.Query
    inputFiles:
      data.csv: "{{outputs.downloadCsv.uri}}"
    sql: |
      SELECT sum(total) as total, avg(quantity) as avg_quantity
      FROM read_csv_auto('{{workingDir}}/data.csv', header=True);
    fetch: true

  - id: slack
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{envs.slack_webhook_reporting}}"
    payload: |
      {"channel": "#reporting", 
      "text": "Current sales numbers: total sales is `${{outputs.analyzeSales.rows[0].total}}` and average sales quantity is `{{outputs.analyzeSales.rows[0].avg_quantity}}`"}

triggers:
  - id: runAfterDWHRefresh
    type: io.kestra.core.models.triggers.types.Flow
    conditions:
      - type: io.kestra.core.models.conditions.types.ExecutionStatusCondition
        in:
          - SUCCESS
      - type: io.kestra.core.models.conditions.types.ExecutionFlowCondition
        namespace: prod
        flowId: dataWarehouseRefresh