anna-geller

id: airbyteDbtCore
namespace: dev

tasks:
  - id: data-ingestion
    type: io.kestra.core.tasks.flows.Parallel
    tasks:
      - id: psyduck
        type: io.kestra.plugin.airbyte.connections.Sync
        connectionId: 4de8ab1e-50ef-4df0-aa01-7f21491081f1

anna-geller

D SELECT order_id, 
>         hash(customer_name) as customer_name_hash, 
>         md5(customer_email) as customer_email_hash, 
>         product_id, 
>         price, 
>         quantity, 
>         total 
> FROM orders;
┌──────────┬──────────────────────┬──────────────────────────────────┬────────────┬────────┬──────────┬─────────┐
│ order_id │  customer_name_hash  │       customer_email_hash        │ product_id │ price  │ quantity │  total  │

anna-geller

id: your_flow_name
namespace: dev

tasks:
  - id: dbt
    type: io.kestra.core.tasks.flows.WorkingDirectory
    tasks:
      - id: cloneRepository
        type: io.kestra.plugin.git.Clone
        url: https://github.com/dbt-labs/jaffle_shop_duckdb

anna-geller

# in-process duckdb
jaffle_shop:
  outputs:
    dev:
      type: duckdb
      path: ':memory:'
      extensions:
        - parquet
  target: dev

anna-geller

CREATE TABLE orders AS 
    SELECT *
    FROM read_csv_auto('https://raw.githubusercontent.com/kestra-io/examples/main/datasets/orders.csv');

SELECT order_id, 
        hash(customer_name) as customer_name_hash, 
        md5(customer_email) as customer_email_hash, 
        product_id, 
        price, 
        quantity, 

anna-geller

curl -o docker-compose.yml https://raw.githubusercontent.com/kestra-io/kestra/develop/docker-compose.yml

anna-geller

CREATE DATABASE orders;
CREATE TABLE orders.main.montly_orders as
    SELECT month(order_date) as order_month, sum(total) as total
    FROM read_csv_auto('s3://kestraio/monthly_orders/*.csv')
    GROUP BY 1
    ORDER BY 2 desc;

anna-geller

SELECT month(order_date) as order_month, sum(total) as total
FROM read_csv_auto('s3://kestraio/monthly_orders/*.csv')
GROUP BY 1
ORDER BY 2 desc;

anna-geller

import boto3
import requests


def extract_and_upload(file: str, bucket: str = "kestraio") -> None:
    url = f"https://raw.githubusercontent.com/kestra-io/datasets/main/{file}"

    response = requests.get(url)
    data = response.content.decode("utf-8")
    s3 = boto3.resource("s3")

anna-geller

id: airbyteDbtCore
namespace: dev

tasks:
  - id: data-ingestion
    type: io.kestra.core.tasks.flows.Parallel
    tasks:
      - id: psyduck
        type: io.kestra.plugin.airbyte.connections.Sync
        connectionId: 4de8ab1e-50ef-4df0-aa01-7f21491081f1