本文へ移動
cccskills
無料GitHub で公開

trino-dbt-platform

Production dbt + Trino platform — dbt-trino adapter profiles.yml (ldap/kerberos/oauth/jwt/certificate auth), all materializations (table/view/incremental/materialized_view/ephemeral), incremental strategies (append/merge/delete+insert), Iceberg table_properties in config blocks (format/partitioning/sorted_by/location), on_schema_change, snapshots with TIMESTAMP(6), slim CI with state:modified+ and --defer, dbt project structure (staging/intermediate/mart), ANALYZE post-hook, session_properties in profiles, multi-thread parallelism

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skillsスター 22(GitHub 22 + cccskills 0)
連携ツールAWS

実行スコア 未検証

インストール方法を見る

含まれるファイル(1)

  • SKILL.md9.4 KB

SKILL.md(原文)

インストールする前に、エージェントに与えられる指示の中身を確認できます。

dbt + Trino Production Platform

When to Use

  • Building a dbt project targeting Trino as the query engine
  • Designing incremental models on Iceberg tables
  • Setting up dbt CI/CD with slim builds using state:modified+
  • Configuring dbt materializations with Iceberg-specific properties
  • Implementing SCD2 snapshots or MERGE-based upserts on Trino/Iceberg

Installation

pip install dbt-trino
# For dbt 1.8+
pip install "dbt-trino>=1.8.0"

profiles.yml

LDAP / Password (most common)

# ~/.dbt/profiles.yml
data_platform:
  target: dev
  outputs:
    dev:
      type: trino
      method: ldap
      host: trino-coordinator.internal
      port: 8080
      database: iceberg          # Trino catalog name
      schema: "{{ env_var('DBT_SCHEMA', 'dev_' ~ env_var('USER', 'local')) }}"
      user: "{{ env_var('TRINO_USER') }}"
      password: "{{ env_var('TRINO_PASSWORD') }}"
      threads: 8
      retries: 3
      session_properties:
        query_max_run_time: 4h
        exchange_compression_codec: LZ4
        join_reordering_strategy: AUTOMATIC
      http_scheme: http

    prod:
      type: trino
      method: ldap
      host: trino-coordinator.internal
      port: 443
      database: iceberg
      schema: gold
      user: "{{ env_var('TRINO_PROD_USER') }}"
      password: "{{ env_var('TRINO_PROD_PASSWORD') }}"
      threads: 16
      retries: 5
      http_scheme: https
      session_properties:
        query_max_run_time: 8h
        spill_enabled: "true"

JWT (service accounts, CI/CD)

    ci:
      type: trino
      method: jwt
      host: trino.internal
      port: 443
      database: iceberg
      schema: ci_dbt
      jwt_token: "{{ env_var('TRINO_JWT_TOKEN') }}"
      threads: 4
      http_scheme: https

OAuth 2.0 (interactive user sessions)

    interactive:
      type: trino
      method: oauth
      host: trino.internal
      port: 443
      database: iceberg
      schema: adhoc
      threads: 4

dbt Project Structure

dbt_project/
├── dbt_project.yml
├── profiles.yml
├── models/
│   ├── staging/             # raw → cleaned 1:1 to sources
│   │   ├── _sources.yml
│   │   ├── stg_orders.sql
│   │   └── stg_customers.sql
│   ├── intermediate/        # business logic joins
│   │   ├── int_order_items.sql
│   │   └── int_customer_orders.sql
│   └── mart/                # final analytical models
│       ├── fact_orders.sql
│       ├── dim_customer.sql
│       └── agg_daily_revenue.sql
├── macros/
│   ├── trino_utils.sql
│   └── generate_schema_name.sql
├── snapshots/
│   └── scd2_customers.sql
└── tests/
# dbt_project.yml
name: data_platform
version: 1.0.0
profile: data_platform

models:
  data_platform:
    staging:
      +materialized: view
      +schema: silver
    intermediate:
      +materialized: ephemeral
    mart:
      +materialized: table
      +on_table_exists: drop      # required for AWS Glue catalog
      +post-hook: "ANALYZE {{ this }}"   # update CBO statistics after every run

vars:
  start_date: '2020-01-01'

Materializations with Iceberg Properties

Table with Iceberg Properties

-- models/mart/fact_orders.sql
{{
  config(
    materialized      = 'table',
    on_table_exists   = 'drop',
    properties = {
      "format":            "'PARQUET'",
      "compression_codec": "'ZSTD'",
      "format_version":    "2",
      "partitioning":      "ARRAY['day(order_date)', 'region']",
      "sorted_by":         "ARRAY['customer_id']",
      "location":          "'s3://data-lake/gold/fact_orders/'"
    },
    post_hook = "ANALYZE {{ this }}"
  )
}}

SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.status,
    o.amount,
    c.region,
    CURRENT_TIMESTAMP AS dbt_updated_at
FROM {{ ref('stg_orders') }} o
JOIN {{ ref('stg_customers') }} c ON o.customer_id = c.customer_id

Incremental: append (streaming/event tables)

-- models/staging/stg_events.sql
{{
  config(
    materialized = 'incremental',
    properties = {
      "format":        "'PARQUET'",
      "partitioning":  "ARRAY['day(event_time)']"
    }
  )
}}

SELECT
    event_id,
    user_id,
    event_type,
    event_time,
    properties
FROM {{ source('kafka_landing', 'events') }}

{% if is_incremental() %}
WHERE event_time > (SELECT MAX(event_time) FROM {{ this }})
{% endif %}

Incremental: merge (upsert/SCD1)

-- models/mart/dim_customer.sql
{{
  config(
    materialized        = 'incremental',
    incremental_strategy = 'merge',
    unique_key          = 'customer_id',
    properties = {
      "format":           "'PARQUET'",
      "format_version":   "2",
      "sorted_by":        "ARRAY['customer_id']"
    }
  )
}}

SELECT
    customer_id,
    name,
    email,
    region,
    tier,
    updated_at
FROM {{ ref('stg_customers') }}

{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

Incremental: delete+insert (partition overwrite)

-- models/mart/fact_daily_revenue.sql
{{
  config(
    materialized         = 'incremental',
    incremental_strategy = 'delete+insert',
    unique_key           = 'order_date',
    properties = {
      "format":        "'PARQUET'",
      "partitioning":  "ARRAY['month(order_date)']"
    }
  )
}}

SELECT
    order_date,
    COUNT(*)               AS order_count,
    SUM(amount)            AS gross_revenue
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
-- Reprocess last 3 days to handle late arrivals
WHERE order_date >= DATE_ADD('day', -3, CURRENT_DATE)
{% endif %}

GROUP BY order_date

Snapshots (SCD2 on Iceberg)

-- snapshots/scd2_customers.sql
{% snapshot scd2_customers %}
{{
  config(
    target_schema = 'snapshots',
    strategy      = 'timestamp',
    unique_key    = 'customer_id',
    updated_at    = 'updated_at',
    properties = {
      "format":         "'PARQUET'",
      "format_version": "2"
    }
  )
}}

SELECT customer_id, name, email, region, tier, updated_at
FROM {{ source('bronze', 'customers') }}

{% endsnapshot %}

Fix Iceberg TIMESTAMP precision (Iceberg requires TIMESTAMP(6) WITH TIME ZONE):

-- macros/trino_utils.sql
{% macro trino__current_timestamp() %}
  current_timestamp(6)
{% endmacro %}

Sources Configuration

# models/staging/_sources.yml
version: 2
sources:
  - name: bronze
    database: iceberg
    schema: bronze
    tables:
      - name: orders
        loaded_at_field: ingested_at
        freshness:
          warn_after: {count: 6, period: hour}
          error_after: {count: 24, period: hour}
      - name: customers
      - name: events

Slim CI: state:modified+ with --defer

# Generate state artifact in production (save to S3)
dbt run --target prod
aws s3 cp target/manifest.json s3://dbt-artifacts/production/manifest.json

# CI pipeline: only run models changed since last production run
aws s3 cp s3://dbt-artifacts/production/manifest.json prod-manifest/

dbt run \
  --target ci \
  --select state:modified+ \
  --defer \
  --state prod-manifest/ \
  --threads 4

dbt test \
  --target ci \
  --select state:modified+ \
  --defer \
  --state prod-manifest/

ANALYZE Post-Hook for CBO

# dbt_project.yml — apply to all mart models
models:
  data_platform:
    mart:
      +post-hook: "ANALYZE {{ this }}"

# For specific high-cardinality columns only
models:
  data_platform:
    mart:
      +post-hook: >
        ANALYZE {{ this }}
        WITH (columns = ARRAY['customer_id', 'order_date', 'region'])

Anti-Patterns

  1. on_table_exists = 'rename' with AWS Glue — Glue does not support table rename; use drop mode instead.
  2. merge strategy without format_version = 2 — Iceberg v1 tables don't support equality delete files needed for MERGE; always set "format_version": "2" in properties.
  3. Not setting sorted_by on large incremental models — unsorted Parquet files disable row-group skipping; add sorted_by on the primary filter column.
  4. Missing ANALYZE post-hook on mart models — fresh tables have no statistics; the optimizer can't choose optimal join strategies without them.
  5. Very wide incremental windows (is_incremental() filter selects all rows) — incremental model that reads the whole table on every run is slower than a full refresh; always bound the watermark filter to a realistic window (e.g., 3 days).
  6. Single thread (threads: 1) in profiles.yml — default is 1; set to 8–16 for parallel model execution to significantly reduce run time.

References

  • dbt-trino adapter: docs.getdbt.com/docs/core/connect-data-platform/trino-setup
  • Trino-specific configs: docs.getdbt.com/reference/resource-configs/trino-configs
  • Related skills: [[trino-iceberg-best-practices]], [[trino-dbt-query-performance]], [[trino-query-optimization]]

レビュー

まだレビューはありません。使ってみた感想をお寄せください。

コメント

ログインなしで書き込めます。使ってみた感想、質問、動かなかった環境などを気軽にどうぞ。

まだコメントはありません。最初のコメントを書いてみませんか。

送信するとすぐに公開されます。メールアドレスや電話番号などの個人情報は書かないでください。URL は 2 つまでです。

0 / 1000 文字

このフォームは reCAPTCHA で保護されており、Google のプライバシーポリシーと利用規約が適用されます。

同じリポジトリのスキル

AIOps autonomous incident response agent — LLM-driven diagnosis loop (Claude tool-use agent with kubectl/SQL/Prometheus tools), alert-to-action pipeline (PagerDuty webhook → agent trigger), automated RCA generation (failure taxonomy + log correlation), self-healing action executor with approval gate, incident severity classification, runbook automation (structured YAML runbooks executed by agent), escalation logic, incident timeline auto-generation, Slack integration for human-in-the-loop approvals

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps capacity planning agent — predictive resource forecasting (Prophet/linear regression on Prometheus metrics), Kubernetes HPA/VPA right-sizing recommendations, node capacity headroom calculator, Kafka partition and broker capacity model, Spark executor sizing from job history, data lake storage growth forecast (S3/GCS), cluster autoscaler simulation, automatic HPA threshold tuning, cost-per-job attribution, capacity planning report generator

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps infrastructure anomaly detection — statistical baseline (z-score/IQR/seasonal decomposition), Prometheus anomaly rules (predict_linear/stddev_over_time/MAD), ML-based detection (Isolation Forest/Prophet/LSTM autoencoder on metrics), metric correlation clustering, automated alert threshold tuning from historical data, Kubernetes resource anomaly detection, data pipeline anomaly signals (row count z-score/freshness drift/throughput drop), Python sklearn/Prophet integration, Grafana anomaly panel

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps observability copilot — natural language to PromQL/LogQL translation, alert explanation in plain English (what fired/why/impact), Grafana dashboard auto-generation from service topology, anomaly narrative generation from metric patterns, on-call context enrichment (related alerts/recent deploys/runbook links), log pattern clustering for noise reduction, SLO status explanation, automated weekly observability health digest

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps continuous platform optimization agent — autonomous optimization loop (observe→analyze→recommend→apply), Kubernetes resource rightsizing (VPA recommendation reader + apply), Kafka consumer lag auto-scale, Spark job configuration tuning from history, idle resource detection (unused deployments/PVCs/topics), compaction and VACUUM scheduling, auto-ANALYZE stale statistics, cost savings attribution, optimization audit trail, human approval gate for high-risk changes

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps autonomous query cost analyzer — Trino system.runtime.query_history cost SQL (CPU time/peak memory/bytes scanned), ClickHouse system.query_log top expensive queries, Spark UI cost attribution by user/job/DAG, BigQuery INFORMATION_SCHEMA.JOBS cost analysis, LLM-driven query rewrite recommendations (missing partition filters/broadcast hints/aggregation pushdown), cost anomaly detection (user over daily budget), automated query tagging, chargeback report by team/project

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

ivanshamaev のスキルをすべて見る

このスキルの問題を報告する