はじめに

データを効率的に処理するうえで BigQuery のようなクラウドデータウェアハウス（以下、DWH）はとても便利なサービスです。

ただ、特に小さな組織や個人では、いろんな事情でクラウド DWH の導入が（まだ）難しい場合もあります。

そんな「データを効率的に運用したいけど、クラウド DWH の導入は難しい」場合に、DuckDB と TROCCO を使って、

将来的にクラウド DWH へ移行することになっても、簡単に移行できるかたちで
できるだけ低コストに

データ基盤を構築し、データを効率的に処理する例を紹介します。

構成

この記事で構築するデータ基盤の構成は以下の図の通りです。

ポイント

TROCCO とクラウドストレージ以外のコンポーネントはシングルノード（手元の PC や適当なサーバー 1 台）上で動かします
- TROCCO はフリープランを使い、その他のコンポーネントは手元の PC 上で動かす場合、必要なコストはクラウドストレージの分だけになります
いろんなシステムに散らばっているデータを TROCCO でクラウドストレージに集めます
- クラウドストレージには TROCCO と DuckDB の両方が対応している AWS S3 か Google Cloud Storage のどちらか（または両方）を使います
クラウドストレージに集めたデータを DuckDB で検索・集計します

さらに、必要なら

クラウドストレージに集めたデータを DuckDB に取り込みます
DuckDB に取り込んだデータを dbt で加工します
DuckDB に取り込んだ（または dbt で加工した）データを Streamlit で可視化したり、その他のツール（i.e. Jupyter Notebook）から使います

ツール・サービス

TROCCO

データを活用するためには、まず、いろんなサービス・システムに散らばっているデータを一箇所に集める必要があります。

だた、この仕組みを自力でつくるには、

サービス・システムごとの仕様を把握する
サービス・システムごとの違いをうまく吸収する

必要があり、とても大変です。

この、データの収集を自動化してくれるクラウドサービスが TROCCO です。

DuckDB

DuckDB は

シングルノードで高速に動作する
クラウドストレージのデータを直接操作できる
ほかのツールと簡単に連携できる

軽量・組込み型の OLAP データベースです。

また、DuckDB は（もちろんデータベースなので、データを管理することもできますが）単にデータをクエリするためのツールやライブラリとして使うこともできます。

dbt（オプション）

dbt はデータを効率的に変換するためのツールです。
dbt について詳しく知りたい場合は以下の記事が参考になります。

Streamlit（オプション）

Streamlit はデータを扱うアプリケーションを簡単につくるためのフレームワークです。
Streamlit について詳しく知りたい場合は以下の記事が参考になります。

例

以下のリポジトリでファイル構成の例を確認できます。

手順

TROCCO でデータを集める

TROCCO を使ってデータを集める（転送する）方法については、すでに詳しく解説してくれている記事がたくさんあるので、この記事では説明を省略します。

以下のような記事を参考に、好きな転送元を選んでデータをクラウドストレージに転送します。

https://zenn.dev/cloud_ace/articles/d8d22e63f86a3b
https://blog.trocco.io/case-study/hubspot_to_bigquery_to_dataportal
https://qiita.com/wakama1994/items/245205c4942cc854dcd4

転送先（クラウドストレージ）の設定については、以下の TROCCO 公式ページに説明があります。

DuckDB をインストールする

DuckDB 公式ページを参考に DuckDB をインストールします。

DuckDB を起動する

DuckDB をインストールしたら、以下のコマンドで DuckDB を起動します。

duckdb default.db

シークレットを作成する

DuckDB を起動したら、DuckDB からクラウドストレージのデータを直接アクセスするため、シークレットを作成します（認証情報を設定します）。

シークレットの作成方法は、クラウドストレージと認証方法の種類によって異なります。
以下の代表的な例と公式ページを参考にシークレットを作成します。

AWS S3 にアクセスキーを使ってアクセスする場合

CREATE OR REPLACE SECRET sample (
    TYPE S3,
    REGION '<AWS リージョン>',
    KEY_ID '<AWS アクセスキー ID>',
    SECRET  '<AWS シークレットアクセスキー>'
);

AWS S3 に設定ファイルの情報でアクセスする場合

CREATE OR REPLACE SECRET sample (
    TYPE S3,
    REGION '<AWS リージョン>',
    PROVIDER CREDENTIAL_CHAIN,
    CHAIN 'config',
    PROFILE '<AWS プロファイル名>'
);

AWS S3 に AWS SSO でアクセスする場合

CREATE OR REPLACE SECRET sample (
    TYPE S3,
    REGION '<AWS リージョン>',
    PROVIDER CREDENTIAL_CHAIN,
    CHAIN 'sso',
    PROFILE '<AWS プロファイル名>'
);

Google Cloud Stroage にアクセスする場合

CREATE OR REPLACE SECRET sample (
    TYPE GCS,
    KEY_ID '<アクセスキー>',
    SECRET  '<シークレット>'
);

DuckDB は AWS S3 互換 API を使って Google Cloud Storage にアクセスします。

クラウドストレージに Google Cloud Storage を使っている場合は、まず Google Cloud 上でアクセスキーとシークレットを作成する必要があります。

アクセスキー・シークレットの作成方法については、以下の記事が参考になります。

クラウドストレージ上のデータを検索する

シークレットを作成したら、DuckDB でデータを検索してみます。

-- AWS S3 上の CSV ファイルからデータをクエリします。
SELECT * FROM READ_CSV('s3://bucket/object.csv');

-- AWS S3 上の CSV ファイルから、まとめてデータをクエリします。
SELECT * FROM READ_CSV('s3://bucket/*.csv');

-- AWS S3 上の Parquet ファイルから、まとめてデータをクエリします。
SELECT * FROM READ_CSV('s3://bucket/*.parquet');

-- AWS S3 上の JSON Lines ファイルからデータをクエリします。
SELECT * FROM READ_JSON_AUTO('s3://bucket/object.jsonl');

-- Google Cloud Storage 上の CSV ファイルからデータをクエリします。
SELECT * FROM READ_CSV('gs://bucket/object.csv');

-- 関数の指定は省略できます。
SELECT * FROM 's3://bucket/object.csv';

-- SELECT も省略できます。
FROM 's3://bucket/object.csv';

ほかにも公式ページでクエリの例を確認できます。

DuckDB にデータを取り込む（オプション）

DuckDB からクラウドストレージ上のファイルにアクセスすると、DuckDB はファイルを毎回ダウンロードします。

同じデータに複数回アクセスしたり、ほかのツールにデータを連携する場合など、ファイルを毎回ダウンロードしたくない場合は、以下のようにクエリ結果を取り込んでテーブルを作成します。

CREATE TABLE sample AS FROM 's3://bucket/object.csv';

データの取込み操作をコード化する

DuckDB にデータを取り込む操作は複雑になったり、複数回繰り返すことがあるため、コード化しておくと運用が楽になります。

コード化には SQL や DuckDB が API を提供している言語が使えます。
たとえば Python なら以下のようなコードになります。

# ingest.py

# このコードは以下のコマンドで実行できます。
#
# pip install duckdb
# python ingest.py

import os
from pathlib import Path

import duckdb


def run():
    with duckdb.connect(database="sample.db") as db:
        db.execute("""
            CREATE OR REPLACE SECRET logs (
                TYPE S3,
                REGION '%(aws_regsion)s',
                KEY_ID '%(aws_s3_access_key_id)s',
                SECRET  '%(aws_s3_secret_access_key)s',
            );

            CREATE OR REPLACE TABLE logs AS FROM 's3://default/logs.csv';
        """ % {
            "aws_region": os.environ["AWS_REGION"],
            "aws_s3_access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
            "aws_s3_secret_access_key": os.environ["AWS_SECRET_ACEESS_KEY"],
        })


if __name__ == "__main__":
    run()

dbt でデータを加工する（オプション）

クラウドストレージ上の（または DuckDB に取り込んだ）データがそのままでは扱いづらい場合、dbt で加工します。

パッケージをインストールする

pip install dbt-core dbt-duckdb

dbt プロジェクトを初期化する

dbt init

profiles.yml を変更する

sample:
  target: "default"
  outputs:
    default:
      type: "duckdb"
      path: "sample.db"
      threads: 4

dbt モデルを作成する

-- models/warnings.sql

SELECT timestamp, message
FROM logs
WHERE level == 'warning'

# models/warnings.yml

version: 2
models:
  - name: "warnings"
    columns:
      - name: "timestmap"
        data_type: "VARCHAR"
      - name: "message"
        data_type: "VARCHAR"

dbt を実行する

dbt run

Streamlit でデータを可視化する（オプション）

クラウドストレージ上のデータや DuckDB に取り込んだ生、または加工済みのデータは Streamlit を使って可視化すると分析しやすくなったり、分析結果を共有しやすくなります。

この記事では Streamlit（と PyGWalker というライブラリ）を使ってデータを可視化します。

パッケージをインストールする

pip install streamlit pygwalker

アプリケーションを作成する

# sample.py

import os
from pathlib import Path

import duckdb
import streamlit
from pygwalker.api.streamlit import StreamlitRenderer


@streamlit.cache_resource
def get_pygwalker_renderer(
    db_name: str,
    db_query: str,
) -> StreamlitRenderer:
    db = duckdb.connect(
        database=Path(os.environ["SD_DATA_DIR"]) / db_name,
        read_only=True,
    )

    records = db.sql(db_query)

    return StreamlitRenderer(records.to_df())


streamlit.set_page_config(
    page_title="Sample",
    layout="wide",
)

streamlit.title("Sample")

streamlit.markdown("""
    This is a sample Streamlit application.
""")

with streamlit.form("Select data"):
    db_name = streamlit.text_input("Database", "sample.db")
    db_query = streamlit.text_area("Query", "SELECT * FROM warnings;")

    if streamlit.form_submit_button("Visualize"):
        pygwalker_renderer = get_pygwalker_renderer(
            db_name=db_name,
            db_query=db_query,
        )

        pygwalker_renderer.explorer()

アプリケーションを起動する

streamlit run sample.py

補足: クラウド DWH への移行

クラウド DWH へ移行したくなったら

クラウドストレージに集めたデータをクラウド DWH からロードする
- クラウド DWH がデータをロードせずに扱う機能（i.e. 外部テーブル）を提供している場合は不要です
TROCCO の dbt 連携で dbt を実行する

よう設定するだけです。

Streamlit やその他のツールについては、

クラウド DWH がサポートしているツールを使う
- たとえば Snowflake には Streamlit を扱う機能があります
クラウド DWH から DuckDB にデータを取り込んで、ツールから DuckDB 経由でデータを使う
ツールからクラウド DWH 上のデータを直接使う

など、適当な方法で連携します。

おわりに

TROCCO と DuckDB を組み合わせることで、クラウド DWH の導入が難しい場合でも、ある程度効率的にデータを運用しつつ、必要になればクラウド DWH へ移行することもできます。

そこまで大げさな話でなくても、クラウドストレージにデータを集めて DuckDB からアクセスするだけでも普段の業務が楽になったりすることもあります。

基本的にはほとんどコストもかからないので、とりあえず気楽にいろいろ試してみてもいいかもしれません。