샤딩

// MongoDB 샤딩 예제 - Node.js
const { MongoClient } = require('mongodb');

/**
 * MongoDB 샤드 클러스터 연결 및 사용
 * mongos (라우터)에 연결하면 자동으로 샤딩 처리됨
 */
class ShardedMongoService {
    constructor(mongosUri) {
        // mongos 라우터에 연결
        this.client = new MongoClient(mongosUri);
        this.db = null;
    }

    async connect() {
        await this.client.connect();
        this.db = this.client.db('ecommerce');
        console.log('Connected to sharded MongoDB cluster');
    }

    /**
     * 샤드 컬렉션 설정
     * 실제로는 mongosh에서 설정하지만, 예시로 포함
     */
    async setupSharding() {
        const adminDb = this.client.db('admin');

        // 데이터베이스 샤딩 활성화
        await adminDb.command({ enableSharding: 'ecommerce' });

        // 컬렉션 샤딩 - user_id를 해시 샤드 키로
        await adminDb.command({
            shardCollection: 'ecommerce.orders',
            key: { user_id: 'hashed' }  // 해시 샤딩
        });

        // Range 샤딩 예시
        await adminDb.command({
            shardCollection: 'ecommerce.products',
            key: { category: 1, product_id: 1 }  // 복합 샤드 키
        });
    }

    /**
     * 주문 생성 - user_id 기반으로 자동 샤드 라우팅
     */
    async createOrder(userId, items, totalAmount) {
        const order = {
            user_id: userId,  // 샤드 키
            items,
            total_amount: totalAmount,
            status: 'pending',
            created_at: new Date()
        };

        // mongos가 user_id 해시값으로 적절한 샤드 선택
        const result = await this.db.collection('orders').insertOne(order);
        console.log(`Order created: ${result.insertedId}`);
        return result.insertedId;
    }

    /**
     * 사용자의 모든 주문 조회
     * 샤드 키(user_id) 포함 쿼리 - 단일 샤드만 접근 (효율적)
     */
    async getOrdersByUser(userId) {
        // user_id가 샤드 키이므로 targeted query
        const orders = await this.db.collection('orders')
            .find({ user_id: userId })
            .sort({ created_at: -1 })
            .toArray();

        console.log(`Found ${orders.length} orders for user ${userId}`);
        return orders;
    }

    /**
     * 날짜 범위 주문 조회
     * 샤드 키 없는 쿼리 - 모든 샤드 스캔 (scatter-gather)
     */
    async getOrdersByDateRange(startDate, endDate) {
        // 샤드 키 없음 - 모든 샤드에서 검색 (비효율적)
        const orders = await this.db.collection('orders')
            .find({
                created_at: {
                    $gte: startDate,
                    $lte: endDate
                }
            })
            .toArray();

        console.log(`Scatter-gather query: ${orders.length} orders found`);
        return orders;
    }

    /**
     * 샤드 분산 상태 확인
     */
    async getShardDistribution() {
        const stats = await this.db.collection('orders').stats();

        console.log('Shard Distribution:');
        console.log(`  Total documents: ${stats.count}`);
        console.log(`  Total size: ${(stats.size / 1024 / 1024).toFixed(2)} MB`);

        // 각 샤드별 청크 정보
        const configDb = this.client.db('config');
        const chunks = await configDb.collection('chunks')
            .find({ ns: 'ecommerce.orders' })
            .toArray();

        const shardCounts = {};
        chunks.forEach(chunk => {
            shardCounts[chunk.shard] = (shardCounts[chunk.shard] || 0) + 1;
        });

        console.log('  Chunks per shard:', shardCounts);
        return shardCounts;
    }

    /**
     * 집계 쿼리 - mapReduce 또는 aggregation
     */
    async getOrderStats() {
        // Aggregation pipeline - 모든 샤드에서 실행 후 mongos가 병합
        const stats = await this.db.collection('orders').aggregate([
            {
                $group: {
                    _id: '$status',
                    count: { $sum: 1 },
                    total_revenue: { $sum: '$total_amount' }
                }
            },
            { $sort: { count: -1 } }
        ]).toArray();

        return stats;
    }
}

// Zone 샤딩 (지역 기반 샤딩) 예시
async function setupZoneSharding(adminDb) {
    // 샤드를 지역 존에 할당
    await adminDb.command({
        addShardToZone: 'shard-seoul',
        zone: 'asia'
    });

    await adminDb.command({
        addShardToZone: 'shard-virginia',
        zone: 'americas'
    });

    // 존 범위 설정 - 지역별로 데이터 배치
    await adminDb.command({
        updateZoneKeyRange: 'ecommerce.users',
        min: { region: 'asia', user_id: MinKey },
        max: { region: 'asia', user_id: MaxKey },
        zone: 'asia'
    });
}

// 사용 예시
async function main() {
    const service = new ShardedMongoService(
        'mongodb://mongos1.example.com:27017,mongos2.example.com:27017'
    );

    await service.connect();

    // 주문 생성 - 자동으로 적절한 샤드에 저장
    await service.createOrder('user_123', [
        { product_id: 'prod_1', quantity: 2 }
    ], 50000);

    // 샤드 분산 상태 확인
    await service.getShardDistribution();
}

main().catch(console.error);

-- Citus (PostgreSQL 샤딩 확장) 예제

-- ============================================
-- 1. 분산 테이블 생성
-- ============================================
-- Citus 확장 활성화
CREATE EXTENSION citus;

-- 워커 노드 추가
SELECT citus_add_node('worker1.db.com', 5432);
SELECT citus_add_node('worker2.db.com', 5432);
SELECT citus_add_node('worker3.db.com', 5432);

-- 분산 테이블 생성 (해시 샤딩)
CREATE TABLE orders (
    order_id BIGSERIAL,
    user_id BIGINT NOT NULL,  -- 샤드 키
    product_id INT NOT NULL,
    quantity INT NOT NULL,
    total_amount DECIMAL(12,2),
    status VARCHAR(20) DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (user_id, order_id)  -- 샤드 키 포함 필수
);

-- user_id를 샤드 키로 분산 (32개 샤드)
SELECT create_distributed_table('orders', 'user_id');


-- ============================================
-- 2. 레퍼런스 테이블 (모든 노드에 복제)
-- ============================================
CREATE TABLE products (
    product_id SERIAL PRIMARY KEY,
    name VARCHAR(200),
    category VARCHAR(50),
    price DECIMAL(10,2)
);

-- 모든 워커에 전체 복제 (조인 성능 향상)
SELECT create_reference_table('products');


-- ============================================
-- 3. 샤드 키 포함 쿼리 (단일 샤드 접근)
-- ============================================
-- 효율적: user_id(샤드 키)로 필터링
SELECT * FROM orders
WHERE user_id = 12345
ORDER BY created_at DESC;

-- 효율적: 샤드 키 포함 조인
SELECT o.*, p.name as product_name
FROM orders o
JOIN products p ON o.product_id = p.product_id
WHERE o.user_id = 12345;


-- ============================================
-- 4. 크로스 샤드 쿼리 (모든 샤드 스캔)
-- ============================================
-- 비효율적: 샤드 키 없는 쿼리 (scatter-gather)
SELECT * FROM orders
WHERE created_at > '2024-01-01';

-- 집계 쿼리 - Citus가 분산 실행 후 병합
SELECT
    DATE_TRUNC('day', created_at) as order_date,
    COUNT(*) as order_count,
    SUM(total_amount) as daily_revenue
FROM orders
WHERE created_at >= '2024-01-01'
GROUP BY DATE_TRUNC('day', created_at)
ORDER BY order_date;


-- ============================================
-- 5. Co-location (같은 샤드 키로 관련 테이블 배치)
-- ============================================
CREATE TABLE order_items (
    item_id BIGSERIAL,
    order_id BIGINT,
    user_id BIGINT NOT NULL,  -- 같은 샤드 키
    product_id INT,
    quantity INT,
    price DECIMAL(10,2),
    PRIMARY KEY (user_id, item_id)
);

-- orders와 같은 샤드에 배치 (Co-location)
SELECT create_distributed_table('order_items', 'user_id',
    colocate_with => 'orders');

-- Co-located 조인 - 로컬 조인으로 매우 효율적
SELECT o.order_id, COUNT(oi.item_id) as item_count
FROM orders o
JOIN order_items oi ON o.order_id = oi.order_id
    AND o.user_id = oi.user_id  -- 샤드 키 조인 필수!
WHERE o.user_id = 12345
GROUP BY o.order_id;


-- ============================================
-- 6. 샤드 밸런싱 및 모니터링
-- ============================================
-- 샤드별 데이터 분포 확인
SELECT
    nodename,
    count(*) as shard_count,
    pg_size_pretty(sum(shard_size)) as total_size
FROM citus_shards
JOIN citus_stat_statements ON shard_id = shardid
GROUP BY nodename;

-- 샤드 리밸런싱
SELECT citus_rebalance_start();

-- 핫스팟 샤드 확인
SELECT
    shardid,
    nodename,
    shard_size,
    total_time_ms
FROM citus_shards s
JOIN (
    SELECT shardid, sum(total_time) as total_time_ms
    FROM citus_stat_statements
    GROUP BY shardid
) q ON s.shardid = q.shardid
ORDER BY total_time_ms DESC
LIMIT 10;


-- ============================================
-- 7. 리샤딩 (샤드 수 변경)
-- ============================================
-- 샤드 수 조정 (다운타임 없이)
SELECT alter_distributed_table('orders', shard_count := 64);