Batch

Run a PostgreSQL batch-query.

yaml
type: "io.kestra.plugin.jdbc.postgresql.Batch"

Examples

Fetch rows from a table, and bulk insert them to another one.

yaml
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( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )

Fetch rows from a table, and bulk insert them to another one, without using sql query.

yaml
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

Use Postgres Batch to bulk insert rows

yaml
  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 (?, ?, ?, ?)

Properties

from *string

Source file URI

Pebble expression referencing an Internal Storage URI e.g. {{ outputs.mytask.uri }}.

url *string

The JDBC URL to connect to the database.

chunk integerstring

Default 1000

The size of chunk for every bulk request.

columns array

SubType string

The columns to be inserted.

If not provided, ? count need to match the from number of columns.

password string

The database user's password.

sql string

Insert query to be executed.

The query must have as many question marks as the number of columns in the table. Example: 'insert into <table_name> values( ? , ? , ? )' for 3 columns. In case you do not want all columns, you need to specify it in the query in the columns property Example: 'insert into <table_name> (id, name) values( ? , ? )' for inserting data into 2 columns: 'id' and 'name'.

ssl booleanstring

Default false

Is the connection SSL?

sslCert string

The SSL cert.

Must be a PEM encoded certificate

sslKey string

The SSL key.

Must be a PEM encoded key

sslKeyPassword string

The SSL key password.

sslMode string

Possible Values

DISABLEALLOWPREFERREQUIREVERIFY_CAVERIFY_FULL

The SSL mode.

sslRootCert string

The SSL root cert.

Must be a PEM encoded certificate

table string

The table from which column names will be retrieved.

This property specifies the table name which will be used to retrieve the columns for the inserted values. You can use it instead of specifying manually the columns in the columns property. In this case, the sql property can also be omitted, an INSERT statement would be generated automatically.

timeZoneId string

The time zone id to use for date/time manipulation. Default value is the worker's default time zone id.

username string

The database user.

Outputs

rowCount integer

The rows count.

updatedCount integer

The updated rows count.

​Batch

Batch