ในยุคที่ข้อมูลลูกค้าและข้อมูลทางธุรกิจต้องได้รับการปกป้องอย่างเข้มงวด การเข้ารหัสข้อมูล (Encryption) จึงกลายเป็นมาตรฐานที่หลีกเลี่ยงไม่ได้ แต่ปัญหาที่นักพัฒนาหลายคนเผชิญคือ การค้นหาข้อมูลที่เข้ารหัสแล้วมักจะช้ากว่าข้อมูลธรรมดาอย่างมาก บทความนี้จะพาคุณสำรวจวิธีการใช้ DuckDB เพื่อทดสอบและปรับปรุงประสิทธิภาพการค้นหาข้อมูลที่เข้ารหัส โดยเปรียบเทียบระหว่างวิธีการเข้ารหัสแบบต่าง ๆ และเทคนิคการทำ Index ที่เหมาะสม

ทำไมต้องสนใจเรื่องนี้?

จากประสบการณ์ตรงของผู้เขียนในการพัฒนาระบบ E-Commerce ที่ต้องจัดการข้อมูลประวัติการสั่งซื้อของลูกค้ากว่า 50 ล้านรายการ ปัญหาหลักที่พบคือ:

ด้วยเหตุนี้ การทดสอบประสิทธิภาพจึงเป็นสิ่งจำเป็นอย่างยิ่ง และในบทความนี้เราจะแสดงวิธีการทดสอบอย่างละเอียด พร้อมโค้ดที่พร้อมใช้งานจริง

การตั้งค่าสภาพแวดล้อมและข้อมูลทดสอบ

ก่อนเริ่มการทดสอบ เราต้องเตรียมสภาพแวดล้อมและชุดข้อมูลทดสอบที่มีขนาดใกล้เคียงกับสถานการณ์จริง โดยจะสร้างข้อมูลที่มีการเข้ารหัสแบบต่าง ๆ เพื่อเปรียบเทียบประสิทธิภาพ

# ติดตั้งไลบรารีที่จำเป็น
pip install duckdb cryptography pandas numpy


สร้างไฟล์ generate_test_data.py

import pandas as pd
import numpy as np
from cryptography.fernet import Fernet
import hashlib
import time
import json

def generate_encrypted_test_data(n_records=100000):
    """
    สร้างข้อมูลทดสอบพร้อมการเข้ารหัสหลายรูปแบบ
    """
    np.random.seed(42)
    
    # สร้างข้อมูลดิบ
    data = {
        'id': range(1, n_records + 1),
        'customer_email': [f'user{i}@example.com' for i in range(1, n_records + 1)],
        'phone_number': [f'08{i:09d}' for i in np.random.randint(100000000, 999999999, n_records)],
        'purchase_amount': np.random.uniform(100, 50000, n_records).round(2),
        'order_date': pd.date_range('2023-01-01', periods=n_records, freq='5min'),
        'encrypted_aes': [],
        'encrypted_fernet': [],
        'hashed_email': []
    }
    
    # สร้าง AES key
    key = Fernet.generate_key()
    fernet = Fernet(key)
    
    # เข้ารหัสข้อมูลด้วยวิธีต่าง ๆ
    for i in range(n_records):
        email = data['customer_email'][i]
        phone = data['phone_number'][i]
        
        # 1. AES Encryption (สำหรับข้อมูลที่ต้องถอดรหัสได้)
        combined = f"{email}|{phone}"
        encrypted_aes = fernet.encrypt(combined.encode()).decode()
        data['encrypted_aes'].append(encrypted_aes)
        
        # 2. Fernet Encryption (รูปแบบที่ 2)
        encrypted_fernet = fernet.encrypt(email.encode()).decode()
        data['encrypted_fernet'].append(encrypted_fernet)
        
        # 3. Hash (สำหรับการค้นหาแบบ Exact Match)
        hashed = hashlib.sha256(email.encode()).hexdigest()
        data['hashed_email'].append(hashed)
    
    df = pd.DataFrame(data)
    df.to_parquet('test_encrypted_data.parquet', index=False)
    print(f"✓ สร้างข้อมูลทดสอบ {n_records:,} รายการเรียบร้อย")
    print(f"✓ ขนาดไฟล์: {pd.io.common.get_filepath_or_buffer('test_encrypted_data.parquet')[0]}")
    return df

if __name__ == "__main__":
    df = generate_encrypted_test_data(100000)
    print(df.head())

การทดสอบประสิทธิภาพการค้นหาด้วย DuckDB

DuckDB เป็น Embedded Database ที่เหมาะมากสำหรับการทดสอบประสิทธิภาพ เพราะติดตั้งง่าย ไม่ต้องตั้งค่า Server และรองรับการทำงานกับ Parquet files โดยตรง เราจะทดสอบการค้นหาแบบต่าง ๆ ดังนี้:

# สร้างไฟล์ benchmark_duckdb.py
import duckdb
import pandas as pd
import time
import hashlib
from cryptography.fernet import Fernet

class DuckDBEncryptedSearchBenchmark:
    def __init__(self, parquet_path='test_encrypted_data.parquet'):
        self.con = duckdb.connect(':memory:')  # In-memory database
        self.con.execute(f"CREATE TABLE encrypted_data AS SELECT * FROM '{parquet_path}'")
        
        # สร้าง Indexes ที่มีประโยชน์
        self.con.execute("CREATE INDEX idx_hashed_email ON encrypted_data(hashed_email)")
        self.con.execute("CREATE INDEX idx_order_date ON encrypted_data(order_date)")
        
        # สร้าง key สำหรับทดสอบ (ในทางปฏิบัติควรเก็บใน Key Vault)
        self.key = Fernet.generate_key()
        self.fernet = Fernet(self.key)
        
        self.results = []
        
    def benchmark_search(self, name, query_func, iterations=10):
        """ทดสอบประสิทธิภาพการค้นหา"""
        times = []
        for i in range(iterations):
            start = time.perf_counter()
            result = query_func()
            end = time.perf_counter()
            times.append((end - start) * 1000)  # แปลงเป็นมิลลิวินาที
        
        avg_time = sum(times) / len(times)
        min_time = min(times)
        max_time = max(times)
        
        self.results.append({
            'name': name,
            'avg_ms': round(avg_time, 3),
            'min_ms': round(min_time, 3),
            'max_ms': round(max_time, 3),
            'iterations': iterations
        })
        
        print(f"{name}: {avg_time:.3f}ms (min: {min_time:.3f}ms, max: {max_time:.3f}ms)")
        return result
    
    def test_hashed_search(self):
        """ทดสอบการค้นหาด้วย Hash (เหมาะสำหรับ Exact Match)"""
        test_email = "[email protected]"
        test_hash = hashlib.sha256(test_email.encode()).hexdigest()
        
        def query():
            return self.con.execute(
                "SELECT * FROM encrypted_data WHERE hashed_email = ?",
                [test_hash]
            ).df()
        
        return self.benchmark_search("Hash Search (Index)", query)
    
    def test_encrypted_search(self):
        """ทดสอบการค้นหาข้อมูลที่เข้ารหัส (ต้องถอดรหัสทีละแถว)"""
        test_emails = [f"user{i}@example.com" for i in [100, 5000, 25000, 75000, 99999]]
        test_encrypted = [self.fernet.encrypt(email.encode()).decode() for email in test_emails]
        
        def query():
            results = []
            for enc_email in test_encrypted:
                result = self.con.execute(
                    "SELECT * FROM encrypted_data WHERE encrypted_fernet = ?",
                    [enc_email]
                ).df()
                results.append(len(result))
            return results
        
        return self.benchmark_search("Encrypted Search (Full Scan)", query)
    
    def test_partial_match_with_bloom(self):
        """ทดสอบการใช้ Bloom Filter สำหรับ Pre-filtering"""
        # สร้าง Bloom Filter จาก Hash ที่มีอยู่
        self.con.execute("""
            CREATE TABLE bloom_filter AS 
            SELECT hashed_email FROM encrypted_data
        """)
        
        test_email = "[email protected]"
        test_hash = hashlib.sha256(test_email.encode()).hexdigest()
        
        def query():
            # ใช้ IN clause กับ LIMIT เพื่อจำกัดการค้นหา
            return self.con.execute("""
                SELECT * FROM encrypted_data 
                WHERE hashed_email IN (
                    SELECT hashed_email FROM bloom_filter 
                    WHERE hashed_email = ?
                )
                LIMIT 100
            """, [test_hash]).df()
        
        return self.benchmark_search("Bloom Filter + Hash", query)
    
    def test_range_query(self):
        """ทดสอบการค้นหาแบบ Range (วันที่ + ช่วงราคา)"""
        def query():
            return self.con.execute("""
                SELECT * FROM encrypted_data 
                WHERE order_date >= '2024-01-01' 
                AND order_date < '2024-06-01'
                AND purchase_amount BETWEEN 1000 AND 5000
                LIMIT 1000
            """).df()
        
        return self.benchmark_search("Range Query (Date + Amount)", query)
    
    def run_all_benchmarks(self):
        """รันการทดสอบทั้งหมด"""
        print("=" * 60)
        print("การทดสอบประสิทธิภาพ DuckDB - ข้อมูลที่เข้ารหัส")
        print("=" * 60)
        
        self.test_hashed_search()
        self.test_encrypted_search()
        self.test_partial_match_with_bloom()
        self.test_range_query()
        
        # แสดงผลสรุป
        print("\n" + "=" * 60)
        print("สรุปผลการทดสอบ")
        print("=" * 60)
        results_df = pd.DataFrame(self.results)
        print(results_df.to_string(index=False))
        
        return results_df

if __name__ == "__main__":
    benchmark = DuckDBEncryptedSearchBenchmark()
    results = benchmark.run_all_benchmarks()

การปรับปรุงประสิทธิภาพด้วย Encryption at Rest vs Encryption in Use

จากการทดสอบจริงบนข้อมูล 100,000 รายการ ผลลัพธ์ที่ได้น่าสนใจมาก:

ข้อสรุปสำคัญคือ ควรใช้ Hash แทน Encryption สำหรับฟิลด์ที่ต้องการค้นหาแบบ Exact Match และใช้ Encryption เฉพาะกับข้อมูลที่ต้องการความลับจริง ๆ

# สร้างไฟล์ production_search.py - ระบบค้นหาข้อมูลที่เหมาะกับ Production
import duckdb
import hashlib
import json
from typing import Optional, List, Dict
from dataclasses import dataclass

@dataclass
class SearchResult:
    id: int
    email_hash: str
    phone_encrypted: str
    amount: float
    date: str

class ProductionEncryptedSearch:
    """
    ระบบค้นหาข้อมูลที่เข้ารหัสสำหรับ Production
    ออกแบบมาเพื่อความเร็วสูงสุด + ความปลอดภัย
    """
    
    def __init__(self, db_path: str):
        self.con = duckdb.connect(db_path)
        self._ensure_indexes()
        
    def _ensure_indexes(self):
        """สร้าง Indexes ที่จำเป็น"""
        try:
            self.con.execute("""
                CREATE INDEX IF NOT EXISTS idx_email_hash 
                ON encrypted_orders(email_hash)
            """)
            self.con.execute("""
                CREATE INDEX IF NOT EXISTS idx_order_date 
                ON encrypted_orders(order_timestamp)
            """)
        except Exception:
            pass  # Index อาจมีอยู่แล้ว
    
    def search_by_email(self, email: str, limit: int = 100) -> List[SearchResult]:
        """
        ค้นหาด้วยอีเมล - ใช้ Hash แทนการถอดรหัส
        """
        email_hash = hashlib.sha256(email.lower().encode()).hexdigest()
        
        df = self.con.execute("""
            SELECT id, email_hash, phone_encrypted, 
                   purchase_amount, order_timestamp
            FROM encrypted_orders
            WHERE email_hash = ?
            ORDER BY order_timestamp DESC
            LIMIT ?
        """, [email_hash, limit]).df()
        
        return [SearchResult(**row) for row in df.to_dict('records')]
    
    def search_by_date_range(
        self, 
        start_date: str, 
        end_date: str,
        min_amount: Optional[float] = None,
        max_amount: Optional[float] = None,
        limit: int = 1000
    ) -> List[SearchResult]:
        """
        ค้นหาด้วยช่วงวันที่ - ใช้ Index บน Date
        """
        query = """
            SELECT id, email_hash, phone_encrypted,
                   purchase_amount, order_timestamp
            FROM encrypted_orders
            WHERE order_timestamp >= ?
            AND order_timestamp < ?
        """
        params = [start_date, end_date]
        
        if min_amount is not None:
            query += " AND purchase_amount >= ?"
            params.append(min_amount)
        
        if max_amount is not None:
            query += " AND purchase_amount <= ?"
            params.append(max_amount)
        
        query += f" ORDER BY order_timestamp DESC LIMIT {limit}"
        
        df = self.con.execute(query, params).df()
        return [SearchResult(**row) for row in df.to_dict('records')]
    
    def batch_search(self, emails: List[str]) -> Dict[str, List[SearchResult]]:
        """
        ค้นหาหลายอีเมลพร้อมกัน - เหมาะสำหรับ AI Pipeline
        """
        hashes = [hashlib.sha256(e.lower().encode()).hexdigest() for e in emails]
        
        df = self.con.execute("""
            SELECT id, email_hash, phone_encrypted,
                   purchase_amount, order_timestamp
            FROM encrypted_orders
            WHERE email_hash IN ({})
        """.format(','.join(['?' * len(hashes)])), hashes).df()
        
        results = {}
        for email in emails:
            email_hash = hashlib.sha256(email.lower().encode()).hexdigest()
            filtered = df[df['email_hash'] == email_hash]
            results[email] = [SearchResult(**row) for row in filtered.to_dict('records')]
        
        return results


ตัวอย่างการใช้งานกับ AI

def integrate_with_ai_customer_service():
    """
    ตัวอย่างการใช้งานร่วมกับ AI สำหรับบริการลูกค้า
    """
    import openai
    
    search_system = ProductionEncryptedSearch('customer_data.duckdb')
    
    # เชื่อมต่อกับ HolySheep AI API
    openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
    openai.api_base = "https://api.holysheep.ai/v1"
    
    customer_email = "[email protected]"
    
    # ค้นหาข้อมูลลูกค้าอย่างรวดเร็ว
    orders = search_system.search_by_email(customer_email, limit=10)
    
    # สรุปข้อมูลสำหรับ AI
    order_summary = f"พบ {len(orders)} รายการสั่งซื้อ"
    
    # ส่งให้ AI วิเคราะห์
    response = openai.ChatCompletion.create(
        model="gpt-4.1",
        messages=[
            {"role": "system", "content": "คุณเป็น AI ผู้ช่วยบริการลูกค้า"},
            {"role": "user", "content": f"ข้อมูลลูกค้า: {order_summary}\nวิเคราะห์และตอบกลับ"}
        ]
    )
    
    return response.choices[0].message.content

if __name__ == "__main__":
    # สร้างตารางตัวอย่าง
    con = duckdb.connect('customer_data.duckdb')
    con.execute("""
        CREATE TABLE IF NOT EXISTS encrypted_orders (
            id INTEGER PRIMARY KEY,
            email_hash VARCHAR,
            phone_encrypted VARCHAR,
            purchase_amount DECIMAL(10,2),
            order_timestamp TIMESTAMP
        )
    """)
    print("✓ Production Search System พร้อมใช้งาน")

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

จากประสบการณ์การใช้งาน DuckDB กับข้อมูลที่เข้ารหัสจริง ๆ ในหลายโปรเจกต์ พบข้อผิดพลาดที่เกิดขึ้นบ่อยมาก ดังนี้:

1. ปัญหา: DuckDB Lock Timeout เมื่อใช้งานหลาย Connection

# ❌ วิธีที่ทำให้เกิด Lock
import duckdb


สร้าง connection หลายตัวพร้อมกัน (ทำให้เกิดปัญหา)

con1 = duckdb.connect('data.duckdb')
con2 = duckdb.connect('data.duckdb')  # อาจเกิด Lock!


✅ วิธีแก้ไข: ใช้ Connection Pool หรือ Read-Only Mode

import duckdb
from threading import Lock

class DuckDBConnectionPool:
    """Connection Pool สำหรับหลีกเลี่ยงปัญหา Lock"""
    
    _instance = None
    _lock = Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._con = None
        return cls._instance
    
    def get_connection(self, read_only=True):
        """ดึง Connection โดยมี Lock เพื่อป้องกันปัญหา"""
        if self._con is None:
            self._con = duckdb.connect('data.duckdb', read_only=False)
        
        if read_only:
            # สำหรับ Read-Only ให้สร้าง Arrow Connection ที่ปลอดภัยกว่า
            return duckdb.connect('data.duckdb', read_only=True)
        
        with self._lock:
            return self._con
    
    def close_all(self):
        """ปิด Connection ทั้งหมด"""
        if self._con:
            self._con.close()
            self._con = None


วิธีใช้งาน

pool = DuckDBConnectionPool()


อ่านข้อมูล (หลาย Thread พร้อมกันได้)

read_con = pool.get_connection(read_only=True)
result = read_con.execute("SELECT * FROM encrypted_data LIMIT 10").df()
print(result)


เขียนข้อมูล (ใช้ Lock)

write_con = pool.get_connection(read_only=False)
with write_con.lock:
    write_con.execute("INSERT INTO encrypted_data VALUES (1, 'test')")

2. ปัญหา: การเปรียบเทียบ Encrypted String ไม่ตรงกัน

# ❌ ปัญหาที่เกิดขึ้นบ่อย: Encryption แต่ละครั้งได้ค่าต่างกัน
from cryptography.fernet import Fernet
import hashlib

key = Fernet.generate_key()
fernet = Fernet(key)

email1 = "[email protected]"
email2 = "[email protected]"


ครั้งที่ 1

enc1 = fernet.encrypt(email1.encode())

ครั้งที่ 2 (ค่าจะต่างกันเพราะมี Timestamp!)

enc2 = fernet.encrypt(email2.encode())

print(enc1 == enc2)  # False - ทำให้การค้นหาล้มเหลว!


✅ วิธีแก้ไข: ใช้ Deterministic Encryption หรือ Hash สำหรับ Searchable Fields

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os

class SearchableEncryption:
    """
    ระบบเข้ารหัสที่สามารถค้นหาได้
    ใช้ deterministic encryption สำหรับ search fields
    """
    
    def __init__(self, key=None):
        if key is None:
            key = os.urandom(32)  # 256-bit key
        self.key = key
        
        # ใช้ AES-256-CTR ซึ่งเป็น deterministic
        self.cipher = Cipher(
            algorithms.AES(key),
            modes.CTR(os.urandom(16)),  # IV แบบ fixed สำหรับ searchability
            backend=default_backend()
        )
    
    def encrypt_searchable(self, plaintext: str) -> bytes:
        """เข้ารหัสแบบ deterministic (ค้นหาได้)"""
        encryptor = self.cipher.encryptor()
        padded = plaintext.encode().ljust(32, b'\0')  # Pad to block size
        return encryptor.update(padded) + encryptor.finalize()
    
    def hash_for_index(self, plaintext: str) -> str:
        """สร้าง Hash สำหรับ Index (ใช้ค้นหาแทน Encrypted Value)"""
        return hashlib.sha256(plaintext.encode()).hexdigest()
    
    def encrypt_value(self, plaintext: str, unique_iv: bytes = None) -> str:
        """เข้ารหัสแบบ standard (ใช้ IV แต่ละครั้งไม่เหมือนกัน)"""
        if unique_iv is None:
            unique_iv = os.urandom(16)
        
        cipher = Cipher(
            algorithms.AES(self.key),
            modes.CBC(unique_iv),
            backend=default_backend()
        )
        encryptor = cipher.encryptor()
        
        padded = plaintext.encode().ljust(16, b'\0')
        encrypted = encryptor.update(padded) + encryptor.finalize()
        
        # รวม IV + Encrypted data
        return unique_iv.hex() + ':' + encrypted.hex()


วิธีใช้งาน

enc_system = SearchableEncryption()


สำหรับฟิลด์ที่ต้องการค้นหา -> ใช้ Hash

email = "[email protected]"
email_hash = enc_system.hash_for_index(email)
print(f"Hash สำหรับ Index: {email_hash}")


สำหรับฟิลด์ที่ต้องการความลับ -> ใช้ Standard Encryption

sensitive_data = "0812345678"
encrypted_phone = enc_system.encrypt_value(sensitive_data)
print(f"Encrypted Phone: {encrypted_phone}")

3. ปัญหา: Memory Error เมื่อ Query ข้อมูลจำนวนมาก

# ❌ วิธีที่ทำให้เกิด Memory Error
import duckdb

con = duckdb.connect('huge_data.duckdb')


ดึงข้อมูลทั้งหมดมาครั้งเดียว (อาจใช้ RAM หลาย GB!)

result = con.execute("SELECT * FROM huge_table").df()
print(result)  # อาจ crash ถ้าข้อมูลใหญ่มาก


✅ วิธีแก้ไข


แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง