Batch

id: postgres_bulk_insert
namespace: company.team

tasks:
  - id: query
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://dev:5432/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT *
      FROM xref
      LIMIT 1500;
    fetchType: STORE

  - id: update
    type: io.kestra.plugin.jdbc.postgresql.Batch
    from: "{{ outputs.query.uri }}"
    url: jdbc:postgresql://prod:5433/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      insert into xref values( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )

id: postgres_bulk_insert
namespace: company.team

tasks:
  - id: query
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://dev:5432/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT *
      FROM xref
      LIMIT 1500;
    fetchType: STORE

  - id: update
    type: io.kestra.plugin.jdbc.postgresql.Batch
    from: "{{ outputs.query.uri }}"
    url: jdbc:postgresql://prod:5433/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    table: xre

  id: postgres_batch
  namespace: company.team

  tasks:
    - id: download_products_csv_file
      type: io.kestra.plugin.core.http.Download
      uri: https://huggingface.co/datasets/kestra/datasets/raw/main/csv/products.csv

    - id: products_csv_to_ion
      type: io.kestra.plugin.serdes.csv.CsvToIon
      from: "{{ outputs.download_products_csv_file.uri }}"

    - id: postgres_create_table
      type: io.kestra.plugin.jdbc.postgresql.Query
      url: "jdbc:postgresql://{{ secret('POSTGRES_HOST') }}:5432/postgres"
      username: "{{ secret('POSTGRES_USERNAME') }}"
      password: "{{ secret('POSTGRES_PASSWORD') }}"
      sql: |
        CREATE TABLE IF NOT EXISTS products(
          product_id varchar(5),
          product_name varchar(100),
          product_category varchar(50),
          brand varchar(50)
        )

    - id: postgres_batch_insert
      type: io.kestra.plugin.jdbc.postgresql.Batch
      url: "jdbc:postgresql://{{ secret('POSTGRES_HOST') }}:5432/postgres"
      username: "{{ secret('POSTGRES_USERNAME') }}"
      password: "{{ secret('POSTGRES_PASSWORD') }}"
      from: "{{ outputs.products_csv_to_ion.uri }}"
      sql: |
        insert into products values (?, ?, ?, ?)