id: snowflake
namespace: company.team

tasks:
  - id: create_database
    type: io.kestra.plugin.jdbc.snowflake.Query
    sql: CREATE OR REPLACE DATABASE kestra;

  - id: create_table
    type: io.kestra.plugin.jdbc.snowflake.Query
    sql: |
      CREATE OR REPLACE TABLE KESTRA.PUBLIC.EMPLOYEES (
        first_name STRING ,
        last_name STRING ,
        email STRING ,
        streetaddress STRING ,
        city STRING ,
        start_date DATE
        );

  - id: extract
    type: io.kestra.plugin.core.http.Download
    uri: https://huggingface.co/datasets/kestra/datasets/raw/main/employees/employees00.csv

  - id: load_to_internal_stage
    type: io.kestra.plugin.jdbc.snowflake.Upload
    from: "{{ outputs.extract.uri }}"
    fileName: employees00.csv
    prefix: raw
    stageName: "@kestra.public.%employees"
    compress: true

  - id: load_from_stage_to_table
    type: io.kestra.plugin.jdbc.snowflake.Query
    sql: >
      COPY INTO KESTRA.PUBLIC.EMPLOYEES FROM @kestra.public.%employees
      FILE_FORMAT = (type = csv field_optionally_enclosed_by='"' skip_header =
      1) PATTERN = '.*employees0[0-9].csv.gz' ON_ERROR = 'skip_file';

  - id: analyze
    type: io.kestra.plugin.jdbc.snowflake.Query
    description: Growth of new hires per month
    sql: >
      SELECT year(START_DATE) as year, monthname(START_DATE) as month, count(*)
      as nr_employees FROM kestra.public.EMPLOYEES GROUP BY year(START_DATE),
      monthname(START_DATE) ORDER BY nr_employees desc;
    fetchType: STORE

  - id: csv_report
    type: io.kestra.plugin.serdes.csv.IonToCsv
    from: "{{ outputs.analyze.uri }}"

pluginDefaults:
  - type: io.kestra.plugin.jdbc.snowflake.Query
    values:
      url: jdbc:snowflake://accountID.snowflakecomputing.com?warehouse=COMPUTE_WH
      username: yourSnowflakeUser
      password: "{{ secret('SNOWFLAKE_PASSWORD') }}"

  - type: io.kestra.plugin.jdbc.snowflake.Upload
    values:
      url: jdbc:snowflake://accountID.snowflakecomputing.com?warehouse=COMPUTE_WH
      username: yourSnowflakeUser
      password: "{{ secret('SNOWFLAKE_PASSWORD') }}"