AI

データパイプライン構築の手順：dltで取込・dbtで変換・Parquetで配信する最小構成とコストの増え方【2026年版】

Adureを利用したインフラ構築

データパイプライン構築で最初に作るべきものは、全社のデータ基盤ではなく、利用者が明日から見る1本の流れです。本記事では、dlt 1.32系でGitHub APIのデータを増分で取り込み、dbt Core 1.12系とDuckDBで整形・テストし、Parquetファイルとして配信するまでを、手元で動くコードで組み立てます。後半では、同じ構成をBigQueryやFivetranへ載せ替えたときに費用が増える箇所を課金単位から試算し、外注やマネージドサービスへ切り替える条件を示します。版番号はいずれも2026年11月時点のものです。

まとめ：データパイプライン構築は出口1本から始め増分取込とテストで費用を抑える

構築の順序は、出口の定義、取込、変換とテスト、配信の4段です。最初の1本は、ノートPC1台で動くdlt・dbt・DuckDBの組み合わせで十分に検証できます。ライセンス費はかかりません。

費用が膨らむのは、全件洗い替えを高い頻度で回したときと、コネクタ製品の課金単位を知らずに接続を増やしたときです。取込は初回から増分にし、変換はテストに通ったものだけを配信してください。書き込みが1プロセスに収まらなくなった時点、またはソースが10本を超えた時点が、DWHやコネクタ製品へ載せ替える目安です。

構築前に決める出口と取込・変換・配信の3段に割り当てるツール構成の選び方

コードを書く前に、どの層に何を置くかを1枚の表にします。設計の原則や処理方式の選び分けはデータパイプラインとはで整理しているので、本記事では層ごとのツールの割り当てに絞ります。

出口の利用者・鮮度・形式の3項目を数字で書き出してから着手する理由

出口とは、データを最後に受け取る人やシステムです。決める項目は3つあります。誰が見るか（例：営業部長がBIで週次に見る）、どの鮮度が必要か（例：前日分が翌朝8時までにそろう）、どの形式で渡すか（例：BIが読めるテーブルかParquetファイル）です。

この3項目が埋まらないうちは構築を始めません。鮮度が「翌朝」で済むなら日次バッチで足り、ストリーミング基盤は要らないと判断できます。形式が決まれば、配信層に置くツールも1つに絞れます。

層ごとに最小構成のOSSと本番で載せ替える先のマネージドを対応させた表

本記事で組む最小構成と、規模が大きくなったときの載せ替え先を対応させると次のとおりです。

層 最小構成（本記事） 本番での載せ替え先の例
取込 dlt 1.32系 Fivetran、Airbyte
保存 DuckDB 1.5系のファイル BigQuery、Snowflake
変換・テスト dbt Core 1.12系＋dbt-duckdb 同じdbtをDWH向けアダプタで実行
配信 Parquetファイル BIからDWHを直接参照
実行制御 cronとシェルスクリプト Airflowなどのオーケストレーター

変換層にdbtを置くのは、保存先を載せ替えても変換のSQLとテストを持ち越せるからです。取込を先に変換してから入れるETLではなく、入れてから変換するELTの形をとっています。両者の違いはETLとはで解説しています。

検証環境を作りdltでGitHub APIの課題データをDuckDBへ増分で取り込む手順

取込の題材には、誰でも認証なしで試せるGitHubの公開リポジトリのIssueを使います。業務では、これを自社SaaSや基幹システムのAPIに置き換えます。

dlt 1.32.0とdbt-duckdb 1.11.1導入とduckdb版固定

PyPIのdltでは2026年10月29日公開の1.32.0が、dbt-duckdbでは10月26日公開の1.11.1が最新です。ここで確認すべきなのは、依存条件と本体の最新版の食い違いです。duckdb本体は10月21日に2.0.0が出ましたが、dltのduckdb extraは依存条件をduckdb<2としています。2つのツールが同じDBファイルを読み書きするので、1系に固定して入れます。

mkdir -p github_pipeline/exports && cd github_pipeline
python -m venv .venv
source .venv/bin/activate
pip install "dlt[duckdb]==1.32.0" "dbt-core==1.12.6" "dbt-duckdb==1.11.1" "duckdb<2"
python -c "import duckdb; print(duckdb.__version__)"

最後の行で1.5系（この時点では1.5.6）が表示されれば準備完了です。DuckDB 2.0は新しいストレージ形式を採用しており、1系からの移行手順はDuckDB 2.0とはにまとめています。DuckDB単体の基本操作はDuckDBの使い方を参照してください。

rest_api_sourceとincrementalのPython増分取込

dltのREST API sourceの公式ドキュメントは、増分取込の書き方として{incremental.start_value}のプレースホルダを推奨しており、paramsの中に増分の定義を直接書く旧来の方式は非推奨と明記しています。その推奨形で書いたのが次のingest_github.pyです。

import dlt
from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {
        "base_url": "https://api.github.com/repos/dlt-hub/dlt/",
        "paginator": {"type": "header_link"},
    },
    "resource_defaults": {
        "primary_key": "id",
        "write_disposition": "merge",
    },
    "resources": [
        {
            "name": "issues",
            "endpoint": {
                "path": "issues",
                "params": {
                    "state": "all",
                    "sort": "updated",
                    "direction": "asc",
                    "per_page": 100,
                    "since": "{incremental.start_value}",
                },
                "incremental": {
                    "cursor_path": "updated_at",
                    "initial_value": "2026-10-01T00:00:00Z",
                },
            },
        },
    ],
})

pipeline = dlt.pipeline(
    pipeline_name="github_issues",
    destination="duckdb",
    dataset_name="raw_github",
)
print(pipeline.run(source))

python ingest_github.pyを実行すると、dltのDuckDB destinationの仕様どおり、作業ディレクトリにgithub_issues.duckdbが作られ、raw_github.issuesテーブルに行が入ります。dltは前回取り込んだupdated_atの最大値を状態として保存し、2回目以降はその値をsinceに入れて更新分だけを取りに行きます。

mergeとprimary_keyで再実行しても行が二重にならない仕組みの確認

同じ期間を2回取り込んでも、行は増えません。write_dispositionをmergeにし、primary_keyにIssueのidを指定しているためです。dltのmerge loadingの解説によると、既定の戦略はdelete-insertで、ステージング用のデータセットに入れてから主キーで重複を除き、宛先の該当行を消して入れ直します。

取り込んだ中身はDuckDBのCLIかPythonで確かめます。入れ子の項目はdltの命名規則により__でつないだ列名に展開されるので、投稿者はuser__login、Pull Requestの情報はpull_request__urlという列になります。

python - <<'EOF'
import duckdb
con = duckdb.connect("github_issues.duckdb", read_only=True)
print(con.sql("select count(*) as rows, count(distinct id) as ids from raw_github.issues"))
print(con.sql("select count(*) from raw_github.issues where pull_request__url is not null"))
EOF

rowsとidsが一致していれば重複はありません。GitHubのIssues APIの公式ドキュメントには、Pull RequestもIssueとして返ると書かれています。2つ目のクエリで数えているのがそのPull Requestの行で、次の変換で取り除きます。

dbtで生データを整形しテストを通ったものだけを配信用Parquetへ書き出す手順

取り込んだ生データは、そのまま利用者に渡しません。dbtで列名と型をそろえ、テストに通ったものだけを出口へ書き出します。dbtのプロジェクト構成の基本はdbt Coreの使い方にまとめています。

profiles.yml・sourcesでdltのDuckDBをdbtから参照

github_pipelineの下にdbt_projectディレクトリを作り、次の3ファイルを置きます。dbt-duckdbのREADMEによると、pathはprofiles.ymlの置き場所からの相対パスで解釈されます。

# dbt_project/dbt_project.yml
name: github_pipeline
version: "1.0.0"
profile: github_pipeline
model-paths: ["models"]

# dbt_project/profiles.yml
github_pipeline:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: "../github_issues.duckdb"

# dbt_project/models/sources.yml
version: 2
sources:
  - name: raw_github
    schema: raw_github
    tables:
      - name: issues

pathを書き忘れると、dbt-duckdbはメモリ上のDBで動き、実行が終わると結果が消えます。dltが作ったファイルを指しているかは、dbt debugで接続が通るかで確かめてください。

Pull Request除外のIssue増分SQLとdata_tests検査

整形用のモデルは、前回より新しい行だけを処理する増分モデルにします。dbtの増分モデルの公式解説の形に沿って、is_incremental()の中で絞り込みます。

-- dbt_project/models/stg_issues.sql
{{ config(materialized='incremental', unique_key='issue_id') }}

select
    id          as issue_id,
    number      as issue_number,
    title,
    state,
    user__login as author,
    created_at,
    updated_at,
    closed_at
from {{ source('raw_github', 'issues') }}
where pull_request__url is null
{% if is_incremental() %}
  and updated_at > (select max(updated_at) from {{ this }})
{% endif %}

テストは別のYAMLに書きます。dbtのdata testsの公式リファレンスでは、テストに渡す引数をargumentsの下に置く形が示されているので、その記法に合わせます。

# dbt_project/models/stg_issues.yml
version: 2
models:
  - name: stg_issues
    columns:
      - name: issue_id
        data_tests:
          - unique
          - not_null
      - name: state
        data_tests:
          - accepted_values:
              arguments:
                values: ["open", "closed"]

IDの重複と欠損、想定外の状態値の3点を検査しています。どのテストをどこまで書くかの判断はdbt testの書き方で、generic・singular・unit testの使い分けとして整理しています。

externalで週次集計をParquetに書き出して配信する設定

出口の形式をParquetファイルにします。この設定では、external materializationによる週次集計の書き出し先と、配信に使うファイル形式を指定します。dbt-duckdbのexternalは、モデルの結果をファイルへ書き出すマテリアライゼーションで、READMEによると形式の既定値はparquetです。例ではdbt_projectディレクトリで実行する前提で、1つ上のexportsに書き出します。

-- dbt_project/models/issue_weekly.sql
{{ config(materialized='external', location='../exports/issue_weekly.parquet') }}

select
    date_trunc('week', created_at) as week_start,
    count(*)                       as opened,
    count(closed_at)               as closed
from {{ ref('stg_issues') }}
group by 1
order by 1

READMEには、externalは増分の戦略に対応していないと書かれています。配信用のファイルは毎回作り直す前提で、集計済みの小さな結果だけを書き出してください。生データ全件をParquetへ毎回出す設計にすると、ファイルの再作成が実行時間の大半を占めます。

dbt buildの実行順序でテスト失敗時に配信ファイルを上書きさせない組み方

取込から配信までを1本のシェルスクリプトでつなぎます。

#!/usr/bin/env bash
# run_pipeline.sh
set -euo pipefail
cd /opt/github_pipeline
source .venv/bin/activate
python ingest_github.py
cd dbt_project
dbt build

dbt buildは、モデルの作成とテストを依存関係に従った順序で実行するコマンドです。dbt buildの公式リファレンスには、上流のテストが失敗すると下流のリソースはSKIPになると明記されています。stg_issuesのテストが落ちた日はissue_weeklyが実行されず、Parquetファイルの内容は前回の正しい結果のままで、上書きは行われません。誤った数字を配るより、古い数字のまま止まるほうが利用者の被害は小さく済みます。

このスクリプトをcronで日次に呼べば、最小構成は完成です。上流の完了を待って起動する仕組みやリトライはデータパイプライン自動化の実装で、止まったことに気づく仕組みはデータパイプライン監視の実装で扱っています。

本番へ移すときのDWH載せ替えと従量課金でコストが増える箇所の試算

最小構成ではライセンス費が0円でした。本番でDWHやコネクタ製品へ載せ替えると、課金単位がそのまま費用の増え方を決めます。

BigQueryとFivetranとOSS構成の課金単位と費用が増えるきっかけの比較表

3つの構成の課金単位を、公式の料金ページから並べると次のとおりです。BigQueryの金額はBigQueryの料金ページの既定表示のリージョンのもので、東京などリージョンにより単価は変わります。

構成 課金単位 費用が増えるきっかけ
BigQuery（オンデマンド） スキャン量：6.25ドル／1TiB 全件スキャンの回数
BigQuery（保存） アクティブ論理ストレージ（時間課金） 生データの保持期間
Fivetran 月間アクティブ行数（MAR）＋接続別基本料 更新される行数と接続数
dlt・dbt・DuckDB ライセンス0円、サーバ代と保守の人件費 保守する人の時間

クエリ料金の無料枠は月1TiBです。アクティブ論理ストレージの保存単価は1GiB・1時間あたり0.000031507ドルで、月10GiBまでが無料枠です。Fivetranの基本料はコネクションごとにかかるため、月間アクティブ行数と接続数を分けて確認してください。

BigQueryの保存料金は1か月を約730時間とすると1GiBあたり月0.023ドル前後で、多くの構成ではクエリ料金のほうが先に効いてきます。料金ページには、クエリごと・参照テーブルごとに最低10MBが課金されること、LIMITを付けてもスキャン量は減らないことも書かれています。LIMIT 10で中身を確かめる習慣は、BigQueryでは費用の削減になりません。BigQuery全体の仕組みと採用判断はBigQueryとはを参照してください。

500GiBのテーブルを毎時全件洗い替えた場合と増分に変えた場合の月額試算

費用の増え方を、上の単価で試算します。500GiB（約0.49TiB）のテーブルを、dbtの全件作り直しで毎時更新するケースです。

  • 毎時の全件洗い替え：0.49TiB×24回×30日で約351TiB。無料枠1TiBを引いて約2,190ドル／月
  • 日次の全件洗い替え：0.49TiB×30回で約14.6TiB。約85ドル／月
  • 毎時の増分（1日分のパーティションだけを読む）：約1.4GiB×720回で約0.98TiB。無料枠に収まる

同じテーブルでも、作り方だけで月額が2,000ドル以上変わります。増分の効果が出るのは、ソースが日付でパーティション分割され、絞り込みの条件がそのパーティション列にかかる場合です。前章のstg_issuesのようにis_incremental()で更新分だけを読む形を、DWHへ載せ替える前から作っておくのがコストダウンの近道です。

FivetranのMARがコネクション単位で数えられることで費用が膨らむ場面

Fivetranの従量課金の公式解説では、MARは暦月の中で同期された重複の無い行数で、主キーで行を識別します。同じ行が月に何度更新されても数えるのは1回です。一方で、MARはコネクションごとに数えられ、月1〜100万MARのコネクションには1本あたり5ドルの基本料がかかります。新しいコネクションは作成から14日間無料です。

費用が膨らむのは2つの場面です。1つは、ソース側の夜間処理が全行のupdated_atを書き換えるテーブルで、毎月ほぼ全行がMARに数えられます。もう1つは、同じソースを開発用と本番用の2つの送信先へ流す構成で、MARも基本料もそれぞれ2倍になる計算です。Fivetranの構成と料金の詳細はFivetranとはで解説しています。

最小構成で本番に出してよい条件と外注・マネージドに切り替える判断基準

ここまでの最小構成は、条件がそろえば本番でもそのまま使えます。載せ替えの判断は、書き込みの競合・ソースの本数・保守する人の3点で決めます。

DuckDBのファイル1つで運用を続けてよい条件と載せ替えが必要になる兆候

DuckDBの同時実行の公式ドキュメントによると、アプリに組み込む形で使う場合、読み書きできるのは1つのプロセスだけで、複数のプロセスから同時に開くときは全員が読み取り専用になります。パイプラインの書き込み中にBIから同じファイルを開く運用は成り立ちません。本記事でParquetへ書き出して配信したのは、この制約を避けるためでもあります。

次の条件がそろう間は、DuckDBのファイル1つで運用を続けて構いません。

  • 書き込むのは日次のパイプライン1本だけで、利用者はParquetか読み取り専用で参照する
  • データ量が1台のサーバのディスクとメモリに収まる
  • 見る人が数人で、同時に重いクエリを投げない

複数のチームが同時に書き込みたい、BIから数十人が直接クエリを投げたい、という要望が出た時点がBigQueryなどのDWHへ移る兆候です。dbtのモデルとテストはアダプタを替えて持ち越せます。

dltを書き続けるよりFivetranやAirbyteのコネクタを買うほうが安い境目

取込の部品は、ソースの種類で判断します。自社の基幹DBや独自のAPIなど、既製のコネクタが無いソースはdltで書くほうが早く確実です。Salesforceや広告媒体のように既製コネクタがあり、API仕様の変更が年に何度も起きるSaaSは、追随を製品に任せたほうが保守の人件費を抑えられます。

境目の目安は、ソースが10本を超えたときです。本数が増えるほど、各SaaSのAPI仕様の変更に追随する作業が担当者の時間を削ります。セルフホストで費用を抑えたいならAirbyteが、運用を手放したいならFivetranが候補です。ソースが2〜3本でdltのコードが数百行に収まっているうちは、製品を入れる必要はありません。

構築を外注するときに社内に残す判断と外部に任せてよい作業の線引き

外部に任せてよいのは、取込コードの実装、dbtのモデルとテストの作成、実行基盤とcronやオーケストレーターの構築です。社内に残すのは、出口の3項目（利用者・鮮度・形式）と、テストで何を異常とみなすかの基準です。accepted_valuesに並べる値や件数のしきい値は業務の定義そのものなので、委託先に決めさせると稼働後の修正がすべて外部経由になります。同じ最小構成をAIの学習データ供給へ広げる手順は、AIデータパイプラインの実装：時点整合・データ版固定・学習と推論の特徴量一致を組む手順で扱っています。

出口の定義から一緒に整理し、本記事のような最小構成で検証してから本番の基盤へ広げたい場合は、データ分析基盤構築・MLOps構築支援で相談を受け付けています。

データパイプライン構築でよくある質問と期間・費用・ツール選びの判断

データパイプラインの構築と導入でよく出る質問に、実装の判断に関わるものから答えます。

データパイプラインはノーコードツールとコードのどちらで構築すべきですか？

ソースが既製コネクタのあるSaaSだけで、変換が列の選択と結合程度ならノーコードのツールで足ります。自社DBや独自APIが含まれる、変換にテストを付けたい、変更履歴をGitで管理したい、のどれかに当てはまるならコードで組んでください。両者は排他ではありません。取込だけを製品に任せ、変換はdbtのコードで持つ構成がよく採られます。

dltとFivetran・Airbyteはどのように使い分ければよいですか？

dltはPythonのライブラリで、既存のスクリプトやオーケストレーターの中に組み込んで使います。サーバを別に立てる必要がありません。Fivetranは管理画面で接続を設定するマネージド製品、AirbyteはOSS版をセルフホストできるコネクタ基盤です。既製コネクタが無いソースはdlt、SaaSが多く運用を手放したいならFivetran、セルフホストで製品の機能が欲しいならAirbyte、という分け方が実務的です。

DuckDBで作った変換処理はBigQueryやSnowflakeへそのまま移せますか？

dbtのモデルとテストの構成はそのまま移せますが、SQLの方言の違いについては、移行先に合わせた手直しが必要です。たとえば本記事のdate_trunc('week', created_at)は、BigQueryではTIMESTAMP_TRUNC(created_at, WEEK)のように引数の順序と関数名が変わります。externalのようなdbt-duckdb固有の設定も、DWH側での置き換え先は通常のテーブルやビューです。移行先が決まっているなら、方言の差を吸収するマクロを最初から使うと手直しが減ります。

データパイプラインのコストダウンはどこから手を付ければよいですか？

最初に見るのは、全件洗い替えをしているモデルと、その実行頻度です。本記事の試算のとおり、毎時の全件作り直しを増分に変えるだけで月額が桁で変わります。次に、使われていない中間テーブルと配信先を棚卸しし、誰も見ていない更新を止めます。コネクタ製品を使っている場合は、全行の更新日時を書き換えるソース側の処理が無いかを確かめてください。MARが膨らむ原因の多くはそこにあります。

構築を外注する前に社内で用意しておくものは何ですか？

出口の3項目（誰が見るか・どの鮮度が必要か・どの形式で渡すか）を数字で書いたメモと、ソースごとの接続方法と認証情報の申請窓口です。接続許可の取得は、実装より時間がかかることが珍しくありません。加えて、正しい数字の見本を1つ用意してください。手作業で集計した先月の売上などがあれば、受け入れ検査でパイプラインの結果と突き合わせられます。進め方の全体像はデータパイプラインとはの構築6ステップを参照してください。

関連記事

お気に入りに入れた記事の一覧

一創の公式 SNS

X では毎朝の新着記事と IT ニュースの速報を、YouTube・TikTok・Instagram・LinkedIn・Facebook では記事の解説動画を配信しています。

この記事は以下の記事からリンクされています

この記事に関連するサービス

資料請求

今日のトレンド記事 直近 24 時間で、いつもより多く読まれている記事

  1. 2026.11.11　テックブログ Microsoft 2026年11月の月例更新：Outlookの悪用済みCVE-2026-124194と修正版の確認方法
  2. 2026.11.11　テックブログ JANISの不正アクセス：最大11,652名の医療機関情報と情報窃取プログラムの検知設定
  3. 2026.11.11　テックブログ SMBC日興証券の不正アクセス：持株会会員最大5.6万人の範囲と受託システムの分離設計
  4. 2026.01.21　コラム 税務調査の録音は禁止？判例・裁決と国税局マニュアルで見る、断られたときの対応
  5. 2026.11.09　テックブログ ステップゴルフ不正アクセス：口座情報の対象確認と偽連絡・引き落としの注意点

RELATED POSTS 関連記事

一覧へ戻る

目次