Analyst DuckDB API

분석가와 Airflow를 위한 DuckDB SQL API 서비스. JSON 데이터를 적재하고 SQL로 조회할 수 있습니다.

환경 구성

환경	도메인	클러스터	네임스페이스	용도
PROD	duckdb.clobe.ai	revenue-data	insighter-prod	운영 데이터 적재/조회
DEV	duckdb.dev.clobe.ai	dev-data-k8s	insighter	개발/테스트

접근 방법

1. 웹 브라우저 (분석가용)

2. Airflow DAG에서 접근 (k8s 내부)

각 환경의 Airflow는 같은 클러스터/네임스페이스에서 실행되므로 내부 서비스 주소를 사용합니다.

Airflow Connection 설정 (Admin > Connections):

Field	Prod	Dev
Conn Id	duckdb_api	duckdb_api
Conn Type	HTTP	HTTP
Host	http://analyst-duckdb.insighter-prod.svc.cluster.local	http://analyst-duckdb.insighter.svc.cluster.local
Port	8000	8000

HttpOperator 사용 예시:

from airflow.providers.http.operators.http import HttpOperator
import json

ingest_task = HttpOperator(
    task_id="push_to_duckdb",
    http_conn_id="duckdb_api",
    endpoint="/api/ingest",
    method="POST",
    headers={"Content-Type": "application/json"},
    data=json.dumps({
        "table": "daily_revenue_stats",
        "data": "{{ ti.xcom_pull(task_ids='bq_aggregate') }}",
        "mode": "replace"
    }),
)

PythonOperator 사용 예시:

import requests

def push_to_duckdb(**ctx):
    data = ctx["ti"].xcom_pull(task_ids="bq_aggregate")
    # prod: insighter-prod / dev: insighter
    resp = requests.post(
        "http://analyst-duckdb.insighter-prod.svc.cluster.local:8000/api/ingest",
        json={"table": "daily_stats", "data": data, "mode": "replace"}
    )
    resp.raise_for_status()
    return resp.json()

3. 외부에서 curl / HTTP 클라이언트

# 데이터 조회 (prod)
curl -X POST https://duckdb.clobe.ai/api/sql \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT * FROM my_table LIMIT 10"}'

# 데이터 적재 (prod)
curl -X POST https://duckdb.clobe.ai/api/ingest \
  -H "Content-Type: application/json" \
  -d '{
    "table": "my_data",
    "data": [{"col1": "val1", "col2": 123}],
    "mode": "replace"
  }'

# 테이블 목록
curl https://duckdb.clobe.ai/api/tables

# 파일 업로드
curl -X POST https://duckdb.clobe.ai/api/upload \
  -F "file=@data.json" -F "table_name=my_data"

API 레퍼런스

Method	Path	설명
POST	/api/sql	SQL 쿼리 실행
POST	/api/ingest	JSON 데이터 적재 (Airflow용)
POST	/api/upload	JSON/CSV 파일 업로드
GET	/api/tables	테이블 목록 + 스키마 조회
DEL	/api/tables/{name}	테이블 삭제

POST /api/sql

SQL 쿼리를 실행하고 결과를 JSON으로 반환합니다.

// Request
{"sql": "SELECT * FROM daily_stats WHERE date >= '2026-02-01'"}

// Response
{
  "columns": ["date", "revenue", "orders"],
  "rows": [["2026-02-20", 1234567, 89], ...],
  "row_count": 20
}

POST /api/ingest

JSON 배열 데이터를 DuckDB 테이블에 적재합니다. Airflow 배치에서 주로 사용합니다.

// Request
{
  "table": "daily_revenue_stats",   // 테이블명 (영숫자+언더스코어)
  "data": [                         // JSON 객체 배열
    {"date": "2026-02-20", "revenue": 1234567, "orders": 89},
    {"date": "2026-02-19", "revenue": 1111111, "orders": 76}
  ],
  "mode": "replace"                 // "replace" 또는 "append" (기본값: append)
}

// Response
{"table": "daily_revenue_stats", "rows_loaded": 2, "total_rows": 2, "mode": "replace"}

replace: 기존 테이블 DROP 후 새로 생성 (매일 전체 교체 시 사용)
append: 기존 테이블에 INSERT (증분 적재 시 사용)

POST /api/upload

JSON 또는 CSV 파일을 업로드하여 테이블을 생성합니다.

curl -X POST http://duckdb.clobe.ai/api/upload \
  -F "file=@report.json" \
  -F "table_name=quarterly_report"

// Response
{"table": "quarterly_report", "rows_loaded": 150, "columns": ["date", "metric", "value"]}

GET /api/tables

모든 테이블의 목록, 행 수, 컬럼 스키마를 반환합니다.

// Response
[
  {
    "name": "daily_stats",
    "row_count": 365,
    "columns": [
      {"name": "date", "type": "VARCHAR"},
      {"name": "revenue", "type": "BIGINT"},
      {"name": "orders", "type": "BIGINT"}
    ]
  }
]

DELETE /api/tables/{name}

지정된 테이블을 삭제합니다.

curl -X DELETE http://duckdb.clobe.ai/api/tables/old_data

// Response
{"deleted": "old_data"}

인프라 정보

항목	Prod	Dev
클러스터	revenue-data	dev-data-k8s
네임스페이스	insighter-prod	insighter
도메인	duckdb.clobe.ai	duckdb.dev.clobe.ai
내부 주소	analyst-duckdb.insighter-prod.svc.cluster.local:8000	analyst-duckdb.insighter.svc.cluster.local:8000
데이터 저장	PersistentVolumeClaim (10Gi, GCE Persistent Disk)
파일 업로드 제한	100MB
인증	없음 (내부 네트워크 접근 전제)