dbt CloudでAmazon Athenaがサポートされたので使ってみた

-- 1. CSVファイルを外部テーブルとして作成(orders)
CREATE EXTERNAL TABLE raw_orders (
    id STRING,
    user_id STRING,
    order_date DATE,
    status STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION 's3://＜CSVファイルが置かれたS3ロケーションURI＞/'
TBLPROPERTIES ('skip.header.line.count'='1');

-- 2. CTASを使用してIcebergテーブルを作成(orders)
CREATE TABLE orders
WITH (
    table_type = 'ICEBERG',
    format = 'parquet',
    write_compression = 'snappy',
    partitioning = ARRAY['order_date'],
    is_external=false,
    location='s3://＜任意のS3ロケーションURI＞/'
)
AS SELECT * FROM raw_orders;

-- 1. CSVファイルを外部テーブルとして作成(customers)
CREATE EXTERNAL TABLE raw_customers (
    id STRING,
    first_name STRING,
    last_name STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION 's3://＜CSVファイルが置かれたS3ロケーションURI＞/'
TBLPROPERTIES ('skip.header.line.count'='1');

-- 2. CTASを使用してIcebergテーブルを作成(customers)
CREATE TABLE customers
WITH (
    table_type = 'ICEBERG',
    format = 'parquet',
    write_compression = 'snappy',
    is_external=false,
    location='s3://＜任意のS3ロケーションURI＞/'
)
AS SELECT * FROM raw_customers;

select
    id as customer_id,
    first_name,
    last_name
from default.customers

select
    id as order_id,
    user_id as customer_id,
    order_date,
    status
from default.orders

{{
 config(
        materialized='incremental',
        table_type='iceberg',
        unique_key='customer_id',
        incremental_strategy='merge'
    )
}}

with stg_customers as (
    select * from {{ ref('stg_customers') }}
),
stg_orders as (
    select * from {{ ref('stg_orders') }}
),
customer_orders as (
    select
        customer_id,
        min(order_date) as first_order_date,
        max(order_date) as most_recent_order_date,
        count(order_id) as number_of_orders
    from stg_orders
    group by 1
),
final as (
    select
        customer_id,
        first_name,
        last_name,
        first_order_date,
        most_recent_order_date,
        coalesce(number_of_orders, 0) as number_of_orders

    from stg_customers
    left join customer_orders using (customer_id)
)
select * from final

{{
 config(
        materialized='incremental',
        table_type='iceberg',
        unique_key='customer_id',
        incremental_strategy='merge'
    )
}}

version: 2

models:
  - name: customers
    description: One record per customer
    columns:
      - name: customer_id
        description: Primary key
        tests:
          - unique
          - not_null
      - name: first_order_date
        description: NULL when a customer has not yet placed an order.

  - name: stg_customers
    description: This model cleans up customer data
    columns:
      - name: customer_id
        description: Primary key
        tests:
          - unique
          - not_null

  - name: stg_orders
    description: This model cleans up order data
    columns:
      - name: order_id
        description: Primary key
        tests:
          - unique
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']
      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id