Tecton

Tecton은 머신러닝을 위한 엔터프라이즈급 Feature Platform입니다. 피처 엔지니어링, 저장, 서빙을 통합 관리하여 ML 모델의 프로덕션 배포와 운영을 가속화합니다. Uber의 Michelangelo를 개발한 팀이 창업했습니다.

핵심 개념

• Feature Store: 피처 정의, 버전 관리, 재사용 가능한 피처 저장소
• Feature Pipeline: 배치, 스트리밍, 실시간 피처 계산 파이프라인
• Feature Serving: 온라인(실시간)과 오프라인(배치) 서빙 지원
• Feature Monitoring: 피처 드리프트, 품질, 신선도 모니터링

Tecton 아키텍처

• Declarative Config: Python SDK로 피처를 코드로 정의 (Feature-as-Code)
• Offline Store: 학습용 히스토리 피처 (S3, BigQuery, Snowflake)
• Online Store: 실시간 추론용 저지연 피처 (DynamoDB, Redis)
• Stream Processing: Spark Structured Streaming, Kafka 연동
• Point-in-Time Correctness: 시점 기준 정확한 피처 조회로 data leakage 방지

주요 이점

Tecton은 Training-Serving Skew를 방지하고, 피처 재사용성을 높여 모델 개발 시간을 단축합니다. 특히 금융(사기 탐지), 이커머스(추천), 라이드셰어(가격 책정) 등 실시간 ML이 중요한 도메인에서 활용됩니다.

Feature 정의 (Tecton SDK)

# features/user_features.py
from tecton import Entity, BatchSource, SnowflakeConfig, batch_feature_view
from datetime import timedelta

# Entity 정의 - 피처가 연결될 비즈니스 객체
user = Entity(
    name="user",
    join_keys=["user_id"],
    description="고객 엔티티"
)

# 데이터 소스 정의
user_transactions = BatchSource(
    name="user_transactions",
    batch_config=SnowflakeConfig(
        database="ANALYTICS",
        schema="PUBLIC",
        table="user_transactions",
        timestamp_field="transaction_time"
    )
)

# 배치 Feature View 정의
@batch_feature_view(
    sources=[user_transactions],
    entities=[user],
    mode="spark_sql",
    batch_schedule=timedelta(days=1),
    ttl=timedelta(days=30),
    online=True,
    offline=True,
    feature_start_time=datetime(2023, 1, 1),
    description="사용자 거래 통계 피처"
)
def user_transaction_features(user_transactions):
    return f"""
        SELECT
            user_id,
            transaction_time,
            COUNT(*) as transaction_count_30d,
            SUM(amount) as total_amount_30d,
            AVG(amount) as avg_transaction_amount,
            MAX(amount) as max_transaction_amount,
            COUNT(DISTINCT merchant_id) as unique_merchants_30d
        FROM {user_transactions}
        WHERE transaction_time >= CURRENT_DATE - INTERVAL '30 days'
        GROUP BY user_id, transaction_time
    """

실시간 스트리밍 Feature

# features/realtime_features.py
from tecton import StreamSource, KafkaConfig, stream_feature_view
from tecton.types import Field, String, Int64, Float64, Timestamp
from datetime import timedelta

# Kafka 스트림 소스
click_events = StreamSource(
    name="click_events",
    stream_config=KafkaConfig(
        bootstrap_servers="kafka:9092",
        topics=["user-clicks"],
        timestamp_field="event_time"
    ),
    schema=[
        Field("user_id", String),
        Field("event_time", Timestamp),
        Field("page_id", String),
        Field("session_id", String)
    ]
)

# 실시간 윈도우 집계 Feature
@stream_feature_view(
    source=click_events,
    entities=[user],
    mode="spark_sql",
    aggregation_interval=timedelta(minutes=1),
    aggregations=[
        Aggregation(column="page_id", function="count", time_windows=[
            timedelta(minutes=5),
            timedelta(minutes=30),
            timedelta(hours=1)
        ]),
        Aggregation(column="session_id", function="count_distinct", time_windows=[
            timedelta(hours=1),
            timedelta(hours=24)
        ])
    ],
    online=True,
    offline=True,
    feature_start_time=datetime(2024, 1, 1),
    description="실시간 사용자 클릭 피처"
)
def user_click_features(click_events):
    return f"""
        SELECT
            user_id,
            event_time,
            page_id,
            session_id
        FROM {click_events}
    """

Feature Service 정의 및 서빙

# feature_services/fraud_detection_service.py
from tecton import FeatureService
from features.user_features import user_transaction_features
from features.realtime_features import user_click_features
from features.device_features import user_device_features

# 여러 Feature View를 묶어 Feature Service 생성
fraud_detection_service = FeatureService(
    name="fraud_detection_service",
    features=[
        user_transaction_features,
        user_click_features,
        user_device_features
    ],
    description="사기 탐지 모델용 피처 서비스"
)

# ----- 추론 시 피처 조회 (Python SDK) -----
import tecton

# Feature Service 가져오기
fs = tecton.get_feature_service("fraud_detection_service")

# 온라인 피처 조회 (실시간 추론)
features = fs.get_online_features(
    join_keys={"user_id": "user_123"}
)

print(features.to_dict())
# {
#     "transaction_count_30d": 45,
#     "total_amount_30d": 12500.0,
#     "avg_transaction_amount": 277.8,
#     "user_click_features__page_id_count_5m": 12,
#     "user_click_features__session_id_count_distinct_1h": 2,
#     ...
# }

# 오프라인 피처 조회 (학습 데이터 생성)
training_events = spark.read.parquet("s3://bucket/training_events/")

training_data = fs.get_historical_features(
    spine=training_events,  # user_id, timestamp 컬럼 포함
    timestamp_key="event_timestamp"
).to_spark()

training_data.write.parquet("s3://bucket/training_features/")

REST API로 피처 서빙

# REST API 호출 예시
import requests

TECTON_API_URL = "https://app.tecton.ai/api/v1/feature-service/get-features"
TECTON_API_KEY = "your-api-key"

headers = {
    "Authorization": f"Tecton-key {TECTON_API_KEY}",
    "Content-Type": "application/json"
}

payload = {
    "params": {
        "feature_service_name": "fraud_detection_service",
        "join_key_map": {
            "user_id": "user_123"
        },
        "workspace_name": "production"
    }
}

response = requests.post(
    TECTON_API_URL,
    headers=headers,
    json=payload
)

features = response.json()["result"]["features"]
print(features)
# [45, 12500.0, 277.8, 12, 2, ...]