별도 서버 없이 Parquet·CSV 수억 건을 SQL로 분석하는 Node.js 백엔드 구축 방법

pnpm add @duckdb/node-api

// src/analytics/duckdb.client.ts
import { DuckDBInstance } from '@duckdb/node-api';
 
async function quickStart() {
  // ':memory:' — 인메모리 DB, 프로세스 종료 시 데이터 사라짐
  const instance = await DuckDBInstance.create(':memory:');
  const conn = await instance.connect();
 
  const result = await conn.run(`
    SELECT 42 AS answer, 'DuckDB works!' AS message
  `);
 
  const rows = await result.fetchAllRows();
  console.log(rows);
  // [{ answer: 42, message: 'DuckDB works!' }]
 
  await conn.close();
  await instance.close();
}
 
quickStart();

const instance = await DuckDBInstance.create('./analytics.duckdb');

// src/analytics/path-validator.ts
import path from 'node:path';
 
export const DATA_BASE_DIR = path.resolve(process.env.DATA_BASE_DIR ?? './data');
 
export function validateDataPath(input: string): string {
  // 영숫자·슬래시·하이픈·언더스코어·점·glob(*) 외 문자 거부
  // 작은따옴표(')를 포함한 SQL 특수문자를 이 단계에서 차단
  if (!/^[\w\-./\*]+$/.test(input)) {
    throw new Error('경로에 허용되지 않은 문자가 포함되어 있습니다');
  }
  const resolved = path.resolve(DATA_BASE_DIR, input);
  if (!resolved.startsWith(DATA_BASE_DIR + path.sep) && resolved !== DATA_BASE_DIR) {
    throw new Error('허용된 디렉터리 외부 경로입니다');
  }
  return resolved;
}

// src/analytics/analytics.service.ts
import { validateDataPath } from './path-validator';
 
async getSalesSummary(year: number) {
  // number 타입이라도 정수·범위 검증
  if (!Number.isInteger(year) || year < 2000 || year > 2100) {
    throw new Error('유효하지 않은 연도입니다');
  }
 
  // glob 패턴으로 여러 Parquet 파일을 한 번에 쿼리
  const safePath = validateDataPath(`sales/${year}/*.parquet`);
 
  const result = await this.conn.run(`
    SELECT
      region,
      SUM(amount)   AS total_sales,
      COUNT(*)      AS order_count,
      AVG(amount)   AS avg_order_value
    FROM '${safePath}'
    WHERE status = 'completed'
    GROUP BY region
    ORDER BY total_sales DESC
  `);
 
  return result.fetchAllRows();
}
 
async analyzeCSV(relativePath: string) {
  const safePath = validateDataPath(relativePath);
 
  const result = await this.conn.run(`
    SELECT *
    FROM read_csv_auto('${safePath}', header = true)
    LIMIT 100
  `);
 
  return result.fetchAllRows();
}

async getTopProducts(categoryId: number, limit: number) {
  const safePath = validateDataPath('orders/*.parquet');
 
  const stmt = await this.conn.prepare(`
    SELECT
      product_id,
      product_name,
      SUM(quantity) AS total_sold
    FROM '${safePath}'
    WHERE category_id = $categoryId
    GROUP BY product_id, product_name
    ORDER BY total_sold DESC
    LIMIT $limit
  `);
 
  try {
    const result = await stmt.run({ categoryId, limit });
    return await result.fetchAllRows();
  } finally {
    await stmt.close();
  }
}

// src/analytics/duckdb.provider.ts
import { DuckDBInstance, DuckDBConnection } from '@duckdb/node-api';
import { ConfigService } from '@nestjs/config';
 
export const DUCKDB_INSTANCE = 'DUCKDB_INSTANCE';
export const DUCKDB_CONNECTION = 'DUCKDB_CONNECTION';
 
export const DuckDBProviders = [
  {
    provide: DUCKDB_INSTANCE,
    useFactory: async (config: ConfigService): Promise<DuckDBInstance> => {
      const dbPath = config.get<string>('DUCKDB_PATH', ':memory:');
      const instance = await DuckDBInstance.create(dbPath);
 
      // threads·memory_limit은 SET 명령으로 설정
      // DuckDBInstance.create의 두 번째 인수 시그니처는 버전마다 다를 수 있으므로
      // 표준 SQL SET 명령이 가장 안전하고 버전에 독립적입니다
      const configConn = await instance.connect();
      try {
        const threads = parseInt(config.get<string>('DUCKDB_THREADS', '4'), 10);
        const memLimit = config.get<string>('DUCKDB_MEMORY_LIMIT', '2GB');
        await configConn.run(`SET threads = ${threads}`);
        await configConn.run(`SET memory_limit = '${memLimit}'`);
      } finally {
        await configConn.close();
      }
 
      return instance;
    },
    inject: [ConfigService],
  },
  {
    provide: DUCKDB_CONNECTION,
    useFactory: async (instance: DuckDBInstance): Promise<DuckDBConnection> => {
      return instance.connect();
    },
    inject: [DUCKDB_INSTANCE],
  },
];

// src/analytics/analytics.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { DuckDBProviders } from './duckdb.provider';
import { AnalyticsService } from './analytics.service';
import { AnalyticsController } from './analytics.controller';
 
@Module({
  imports: [ConfigModule],
  providers: [...DuckDBProviders, AnalyticsService],
  controllers: [AnalyticsController],
  exports: [AnalyticsService],
})
export class AnalyticsModule {}

// src/analytics/analytics.service.ts
import { Injectable, Inject, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { DuckDBInstance, DuckDBConnection } from '@duckdb/node-api';
import { DUCKDB_INSTANCE, DUCKDB_CONNECTION } from './duckdb.provider';
import { validateDataPath } from './path-validator';
 
@Injectable()
export class AnalyticsService implements OnModuleInit, OnModuleDestroy {
  constructor(
    @Inject(DUCKDB_INSTANCE) private readonly instance: DuckDBInstance,
    @Inject(DUCKDB_CONNECTION) private readonly conn: DuckDBConnection,
  ) {}
 
  async onModuleInit() {
    // httpfs 확장은 모듈 초기화 시 한 번만 로드
    // INSTALL은 최초 실행 시에만 실제 다운로드가 발생하고 이후는 no-op
    await this.conn.run(`INSTALL httpfs; LOAD httpfs;`);
 
    // S3 자격증명: CREDENTIAL_CHAIN이 환경변수·IAM Role·인스턴스 프로파일을
    // 순서대로 탐색하므로 코드에 키를 노출하지 않아도 됩니다
    await this.conn.run(`
      CREATE OR REPLACE SECRET aws_s3 (
        TYPE S3,
        PROVIDER CREDENTIAL_CHAIN
      );
    `);
  }
 
  async onModuleDestroy() {
    try {
      await this.conn.close();
    } finally {
      await this.instance.close();
    }
  }
 
  async getDailySummary(date: string) {
    // YYYY-MM-DD 형식 검증
    if (!/^\d{4}-\d{2}-\d{2}$/.test(date)) {
      throw new Error('유효하지 않은 날짜 형식입니다');
    }
 
    const safePath = validateDataPath(`events/${date}/*.parquet`);
    const stmt = await this.conn.prepare(`
      SELECT
        DATE_TRUNC('hour', event_time) AS hour,
        event_type,
        COUNT(*) AS event_count
      FROM '${safePath}'
      WHERE event_time >= $startDate::TIMESTAMP
        AND event_time < $startDate::TIMESTAMP + INTERVAL 1 DAY
      GROUP BY 1, 2
      ORDER BY 1, 2
    `);
 
    try {
      const result = await stmt.run({ startDate: date });
      return await result.fetchAllRows();
    } finally {
      await stmt.close();
    }
  }
}

async queryS3Parquet(s3Prefix: string) {
  // s3Prefix 예: 'logs/2026/07'
  // S3 경로는 별도 허용 prefix 검증 권장
  if (!/^[\w\-./]+$/.test(s3Prefix)) {
    throw new Error('유효하지 않은 S3 경로입니다');
  }
 
  const result = await this.conn.run(`
    SELECT
      user_id,
      COUNT(*) AS page_views,
      COUNT(DISTINCT session_id) AS sessions
    FROM 's3://my-analytics-bucket/${s3Prefix}/*.parquet'
    WHERE event_type = 'page_view'
    GROUP BY user_id
    HAVING page_views > 100
    ORDER BY page_views DESC
    LIMIT 1000
  `);
 
  return result.fetchAllRows();
}

async *streamResultByPage(
  relativePath: string,
  pageSize = 10_000,
): AsyncGenerator<Record<string, unknown>[]> {
  const safePath = validateDataPath(relativePath);
  let offset = 0;
 
  while (true) {
    const stmt = await this.conn.prepare(`
      SELECT * FROM '${safePath}'
      LIMIT $pageSize OFFSET $offset
    `);
 
    try {
      const result = await stmt.run({ pageSize, offset });
      const rows = await result.fetchAllRows() as Record<string, unknown>[];
      if (rows.length === 0) break;
      yield rows;
      if (rows.length < pageSize) break;
      offset += pageSize;
    } finally {
      await stmt.close();
    }
  }
}

for await (const page of service.streamResultByPage('large-dataset.parquet')) {
  await processBatch(page);
}

async convertCSVToParquet(csvRelPath: string, outRelPath: string) {
  const csvPath = validateDataPath(csvRelPath);
  const outPath = validateDataPath(outRelPath);
 
  await this.conn.run(`
    COPY (
      SELECT * FROM read_csv_auto('${csvPath}', header = true)
    ) TO '${outPath}'
    (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000)
  `);
}

// 위험 — 경로 탈출도, SQL 특수문자도 차단 못함
await conn.run(`SELECT * FROM '${userInput}'`);
 
// 잘못된 방향 — path.basename은 디렉터리 탈출만 막고
// userInput에 ' 문자가 있으면 SQL이 여전히 깨집니다
const name = path.basename(userInput);
await conn.run(`SELECT * FROM './data/${name}'`);
 
// 올바른 방향 — validateDataPath로 허용 문자 + 디렉터리 경계 동시 검증
const safePath = validateDataPath(userInput);
await conn.run(`SELECT * FROM '${safePath}'`);