Register S3 JSON files as governed datasets and curate them with DuckDB

id: s3-dataset-assets
namespace: company.team
description: Pull JSON from S3, unify into a dataset, stage to S3, curate with
  DuckDB, and ship governed asset lineage.

inputs:
  - id: bucket
    type: STRING
    defaults: "my-json-data-bucket"
    description: S3 bucket containing the JSON files to ingest.
  - id: regexp
    type: STRING
    defaults: "^lead-.*\\.json$"
    description: Regex filter for the JSON files to include.
  - id: region
    type: STRING
    defaults: "eu-central-1"
    description: AWS region for the source bucket.

tasks:
  - id: download_json_from_s3
    type: io.kestra.plugin.aws.s3.Downloads
    description: Download JSON files from S3 and register the raw files as dataset assets.
    bucket: "{{ inputs.bucket }}"
    regexp: "{{ inputs.regexp }}"
    region: "{{ inputs.region }}"
    action: NONE
    assets:
      outputs:
        - id: raw_json_files
          type: io.kestra.plugin.ee.assets.Dataset
          namespace: "{{ flow.namespace }}"
          metadata:
            location: "s3://{{ inputs.bucket }}"
            format: json
            layer: raw

  - id: concat_dynamic_one
    type: io.kestra.plugin.core.storage.Concat
    description: Merge the downloaded JSON files into a single staging dataset.
    files: "{{ outputs.download_json_from_s3.outputFiles | jq('.[]') }}"
    assets:
      inputs:
        - id: raw_json_files
          type: io.kestra.plugin.ee.assets.Dataset
          namespace: "{{ flow.namespace }}"
      outputs:
        - id: unified_json_dataset
          type: io.kestra.plugin.ee.assets.Dataset
          namespace: "{{ flow.namespace }}"
          metadata:
            location: "{{ workingDir }}/unified-data.json"
            format: json
            layer: staging

  - id: upload_unified_json_to_s3
    type: io.kestra.plugin.aws.s3.Upload
    description: Upload the unified JSON file to S3.
    bucket: "{{ inputs.bucket }}"
    key: "processed/{{ flow.id }}/unified-data.json"
    from: "{{ outputs.concat_dynamic_one.uri }}"
    assets:
      inputs:
        - id: unified_json_dataset
          type: io.kestra.plugin.ee.assets.Dataset
          namespace: "{{ flow.namespace }}"
      outputs:
        - id: unified_json_dataset_s3
          type: io.kestra.plugin.ee.assets.Dataset
          namespace: "{{ flow.namespace }}"
          metadata:
            location: "s3://{{ inputs.bucket }}/processed/{{ flow.id }}/unified-data.json"
            format: json
            layer: staging_s3

  - id: duckdb_process_and_query
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Load the unified JSON into DuckDB, create a curated table, and
      persist as a dataset asset.
    inputFiles:
      unified-data.json: "{{ outputs.concat_dynamic_one.uri }}"
    sql: |
      CREATE TABLE IF NOT EXISTS unified_dataset AS
      SELECT * FROM read_json('unified-data.json');
      INSERT INTO unified_dataset SELECT * FROM read_json('unified-data.json');
      SELECT * FROM unified_dataset;
    fetchType: STORE
    outputDbFile: true
    assets:
      inputs:
        - id: unified_json_dataset
          type: io.kestra.plugin.ee.assets.Dataset
          namespace: "{{ flow.namespace }}"
      outputs:
        - id: unified_json_dataset
          type: io.kestra.plugin.ee.assets.Dataset
          namespace: "{{ flow.namespace }}"
          metadata:
            location: "{{ workingDir }}/unified-leads.json"
            table: unified_dataset
            format: json
            layer: curated

  - id: ship_assets
    type: io.kestra.plugin.ee.assets.AssetShipper
    description: Export the curated dataset asset for downstream consumers.
    assetExporters:
      - id: file_exporter
        type: io.kestra.plugin.ee.assets.FileAssetExporter
        format: JSON

pluginDefaults:
  - type: io.kestra.plugin.aws
    values:
      accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
      secretKeyId: "{{ secret('AWS_SECRET_KEY_ID') }}"
      region: "{{ inputs.region }}"
      allowFailure: false