Developer's Catalog

データベース

DuckDB + Vector Search

概要

DuckDBは、インプロセスSQLのOLAP（分析処理）データベースで、SQLiteのような埋め込み型でありながら、分析ワークロードに最適化されています。VSS（Vector Similarity Search）拡張機能により、ベクトル検索機能が追加され、ローカルでの高速なベクトル検索と分析クエリの組み合わせが可能になります。

詳細

DuckDBは2018年にCWI（Centrum Wiskunde & Informatica）で開発が始まり、現在はDuckDB Labsが主導しています。カラムナ型ストレージエンジンと高度なクエリ最適化により、分析クエリで優れたパフォーマンスを発揮します。VSS拡張機能により、ベクトル検索機能が追加され、データ分析とAIワークロードを統合できます。

DuckDB + ベクトル検索の主な特徴：

• 埋め込み型データベースでのベクトル検索
• HNSW（Hierarchical Navigable Small World）インデックス
• 高速な分析クエリとの統合
• メモリ効率的な処理
• パラレル処理とベクトル化実行
• Python、R、Javaなどへの直接統合
• ゼロコピーデータ共有
• Parquet、CSV、JSONなどの直接読み込み
• SQLによる簡単な操作
• 軽量（数MB）のバイナリ

アーキテクチャの特徴

• カラムナ型ストレージ
• ベクトル化クエリ実行エンジン
• 並列処理とマルチコア最適化
• プッシュダウン最適化

メリット・デメリット

メリット

• 簡単なデプロイ: サーバー不要の埋め込み型データベース
• 高速な分析: OLAP最適化による優れたパフォーマンス
• 低レイテンシ: インプロセス実行による最小オーバーヘッド
• 豊富な機能: 完全なSQL機能とウィンドウ関数
• 統合の容易さ: Python、R、Javaへの直接バインディング
• コスト効率: オープンソースで無料

デメリット

• スケーラビリティ: 単一マシンに制限
• 同時実行性: 書き込みは単一プロセスのみ
• 永続性: プライマリストレージとしては不適切
• ベクトル機能の制限: 専用ベクトルDBと比較して機能が限定的
• メモリ制限: 大規模データセットではメモリ不足の可能性

主要リンク

• 公式サイト
• ドキュメント
• GitHub
• VSS拡張機能

書き方の例

セットアップとインストール

import duckdb
import numpy as np
import pandas as pd

# DuckDBの接続とVSS拡張機能のインストール
conn = duckdb.connect(':memory:')

# VSS拡張機能のインストール
conn.execute("INSTALL vss;")
conn.execute("LOAD vss;")

# テーブルの作成
conn.execute("""
    CREATE TABLE documents (
        id INTEGER PRIMARY KEY,
        title VARCHAR,
        content TEXT,
        embedding FLOAT[768],
        category VARCHAR,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

ベクトルデータの挿入と検索

# ドキュメントの挿入
def insert_document(conn, title, content, embedding):
    conn.execute("""
        INSERT INTO documents (title, content, embedding, category)
        VALUES (?, ?, ?, ?)
    """, [title, content, embedding.tolist(), 'technology'])

# サンプルデータの挿入
embedding = np.random.rand(768).astype(np.float32)
insert_document(
    conn,
    "DuckDBベクトル検索",
    "埋め込み型データベースでの高速ベクトル検索",
    embedding
)

# HNSWインデックスの作成
conn.execute("""
    CREATE INDEX idx_embedding ON documents 
    USING HNSW (embedding) 
    WITH (metric = 'cosine', ef_construction = 128, M = 16)
""")

# ベクトル検索
def vector_search(conn, query_vector, limit=10):
    result = conn.execute("""
        SELECT 
            id, 
            title, 
            content,
            array_cosine_similarity(embedding, ?::FLOAT[768]) as similarity
        FROM documents
        ORDER BY array_cosine_similarity(embedding, ?::FLOAT[768]) DESC
        LIMIT ?
    """, [query_vector.tolist(), query_vector.tolist(), limit])
    
    return result.fetchdf()

# 検索実行
query_embedding = np.random.rand(768).astype(np.float32)
results = vector_search(conn, query_embedding)
print(results)

分析クエリとの統合

# 分析データの準備
conn.execute("""
    CREATE TABLE analytics_data AS
    SELECT 
        d.*,
        EXTRACT(MONTH FROM created_at) as month,
        LENGTH(content) as content_length
    FROM documents d
""")

# ベクトル検索と分析の組み合わせ
def analytical_vector_search(conn, query_vector, category=None):
    query = """
        WITH vector_results AS (
            SELECT 
                id,
                title,
                category,
                array_cosine_similarity(embedding, ?::FLOAT[768]) as similarity
            FROM documents
            WHERE array_cosine_similarity(embedding, ?::FLOAT[768]) > 0.8
        )
        SELECT 
            vr.category,
            COUNT(*) as count,
            AVG(vr.similarity) as avg_similarity,
            MIN(vr.similarity) as min_similarity,
            MAX(vr.similarity) as max_similarity
        FROM vector_results vr
        GROUP BY vr.category
        ORDER BY avg_similarity DESC
    """
    
    return conn.execute(
        query, 
        [query_vector.tolist(), query_vector.tolist()]
    ).fetchdf()

# ウィンドウ関数を使用した高度な分析
def trending_similar_documents(conn, query_vector):
    query = """
        WITH ranked_docs AS (
            SELECT 
                id,
                title,
                created_at,
                array_cosine_similarity(embedding, ?::FLOAT[768]) as similarity,
                ROW_NUMBER() OVER (
                    PARTITION BY DATE_TRUNC('day', created_at) 
                    ORDER BY array_cosine_similarity(embedding, ?::FLOAT[768]) DESC
                ) as daily_rank
            FROM documents
            WHERE created_at >= CURRENT_DATE - INTERVAL 30 DAY
        )
        SELECT 
            DATE_TRUNC('day', created_at) as date,
            title,
            similarity,
            daily_rank
        FROM ranked_docs
        WHERE daily_rank <= 5
        ORDER BY date DESC, daily_rank
    """
    
    return conn.execute(
        query,
        [query_vector.tolist(), query_vector.tolist()]
    ).fetchdf()

バッチ処理と最適化

# バッチ挿入（Pandas DataFrameから）
def batch_insert_from_dataframe(conn, df):
    # DataFrameから直接DuckDBテーブルを作成
    conn.register('temp_df', df)
    
    conn.execute("""
        INSERT INTO documents (title, content, embedding, category)
        SELECT 
            title,
            content,
            embedding::FLOAT[768],
            category
        FROM temp_df
    """)

# Parquetファイルから直接ベクトル検索
def search_from_parquet(conn, parquet_file, query_vector):
    query = """
        SELECT 
            title,
            content,
            array_cosine_similarity(embedding::FLOAT[768], ?::FLOAT[768]) as similarity
        FROM read_parquet(?)
        WHERE array_cosine_similarity(embedding::FLOAT[768], ?::FLOAT[768]) > 0.7
        ORDER BY similarity DESC
        LIMIT 10
    """
    
    return conn.execute(
        query,
        [query_vector.tolist(), parquet_file, query_vector.tolist()]
    ).fetchdf()

# メモリ使用量の最適化
def optimize_memory_usage(conn):
    # メモリ制限の設定
    conn.execute("SET memory_limit='2GB';")
    
    # 一時ファイルディレクトリの設定
    conn.execute("SET temp_directory='/tmp/duckdb';")
    
    # ワーカースレッド数の設定
    conn.execute("SET threads=4;")