Source
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 (?, ?, ?, ?)
About this blueprint
Postgres
This flow demonstrates using Postgres Batch task to bulk insert rows.
The flow has the following tasks:
- download_products_csv_file: This task downloads the CSV file containing products data using HTTP download.
- products_csv_to_ion: This task converts the products data from CSV format into ION format.
- postgres_create_table: This task creates the
productstable in Postgres if it does not exists. - postgres_batch_insert: This task bulk inserts the product data rows into Postgres
productstable using Batch task.
More Related Blueprints