Triton

Triton

개요 (Overview)

Triton 모델 함수(Model Function)는 Flink SQL이 NVIDIA Triton Inference Server를 호출해 실시간 모델 추론 작업을 수행할 수 있게 해줍니다. Triton Inference Server는 TensorFlow, PyTorch, ONNX, TensorRT 등 여러 머신러닝 프레임워크를 지원하는 고성능 추론 서빙 솔루션입니다.

주요 기능:

• 고성능 (High Performance): 저지연·고처리량 추론에 최적화됨
• 다중 프레임워크 지원 (Multi-Framework Support): 다양한 ML 프레임워크의 모델과 동작함
• 비동기 처리 (Asynchronous Processing): 더 나은 리소스 활용을 위한 비블로킹 추론 요청
• 유연한 구성 (Flexible Configuration): 다양한 사용 사례를 위한 포괄적인 구성 옵션
• 리소스 관리 (Resource Management): 효율적인 HTTP 클라이언트 풀링과 자동 리소스 정리
• 장애 허용 (Fault Tolerance): 구성 가능한 시도 횟수를 가진 내장 재시도 메커니즘

flink-model-triton 모듈은 Flink 2.3부터 사용할 수 있습니다. 실행 중인 Triton Inference Server 인스턴스에 접근할 수 있는지 확인하세요.

출처: 문서

본문

사용 예시 (Usage Examples)

예시 1: 텍스트 분류 (기본) (Example 1: Text Classification (Basic))

이 예시는 영화 리뷰에 대한 감성 분석을 보여줍니다.

SQL

-- Create the Triton model
CREATE MODEL triton_sentiment_classifier
INPUT (`input` STRING)
OUTPUT (`output` STRING)
WITH (
    'provider' = 'triton',
    'endpoint' = 'http://localhost:8000/v2/models',
    'model-name' = 'text-classification',
    'model-version' = '1',
    'timeout' = '10000'
);

-- Prepare source data
CREATE TEMPORARY VIEW movie_reviews(id, movie_name, user_review, actual_sentiment)
AS VALUES
  (1, 'Great Movie', 'This movie was absolutely fantastic! Great acting and storyline.', 'positive'),
  (2, 'Boring Film', 'I fell asleep halfway through. Very disappointing.', 'negative'),
  (3, 'Average Show', 'It was okay, nothing special but not terrible either.', 'neutral');

-- Create output table
CREATE TEMPORARY TABLE classified_reviews(
  id BIGINT,
  movie_name VARCHAR,
  predicted_sentiment VARCHAR,
  actual_sentiment VARCHAR
) WITH (
  'connector' = 'print'
);

-- Classify sentiment
INSERT INTO classified_reviews
SELECT id, movie_name, output as predicted_sentiment, actual_sentiment
FROM ML_PREDICT(
  TABLE movie_reviews,
  MODEL triton_sentiment_classifier,
  DESCRIPTOR(user_review)
);

Table API (Java)

TableEnvironment tEnv = TableEnvironment.create(...);

// Register the model
tEnv.executeSql(
    "CREATE MODEL triton_sentiment_classifier " +
    "INPUT (`input` STRING) " +
    "OUTPUT (`output` STRING) " +
    "WITH (" +
    "  'provider' = 'triton', " +
    "  'endpoint' = 'http://localhost:8000/v2/models', " +
    "  'model-name' = 'text-classification', " +
    "  'model-version' = '1', " +
    "  'timeout' = '10000'" +
    ")"
);

// Register source table
tEnv.executeSql(
    "CREATE TEMPORARY VIEW movie_reviews(id, movie_name, user_review, actual_sentiment) " +
    "AS VALUES " +
    "  (1, 'Great Movie', 'This movie was absolutely fantastic!', 'positive'), " +
    "  (2, 'Boring Film', 'I fell asleep halfway through.', 'negative')"
);

// Perform classification
Table result = tEnv.sqlQuery(
    "SELECT id, movie_name, output as predicted_sentiment " +
    "FROM ML_PREDICT(" +
    "  TABLE movie_reviews, " +
    "  MODEL triton_sentiment_classifier, " +
    "  DESCRIPTOR(user_review)" +
    ")"
);

result.execute().print();

예시 2: 스트리밍을 이용한 이미지 분류 (Example 2: Image Classification with Streaming)

ResNet 모델로 Kafka 스트림의 이미지를 분류합니다:

-- Register image classification model
CREATE MODEL image_classifier
INPUT (image_pixels ARRAY<FLOAT>)
OUTPUT (predicted_class STRING, confidence FLOAT)
WITH (
  'provider' = 'triton',
  'endpoint' = 'http://triton-server:8000/v2/models',
  'model-name' = 'resnet50',
  'model-version' = '1',
  'timeout' = '10000',
  'compression' = 'gzip'  -- Enable compression for large image data
);

-- Source table from Kafka
CREATE TEMPORARY TABLE image_stream (
  image_id STRING,
  image_pixels ARRAY<FLOAT>,  -- Preprocessed image as float array
  upload_time TIMESTAMP(3),
  WATERMARK FOR upload_time AS upload_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'images',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

-- Classify images
SELECT
  image_id,
  predicted_class,
  confidence,
  upload_time
FROM ML_PREDICT(
  TABLE image_stream,
  MODEL image_classifier,
  DESCRIPTOR(image_pixels)
);

예시 3: 실시간 사기 탐지 (Example 3: Real-time Fraud Detection)

사기 탐지를 위한 고우선순위 추론:

-- Create fraud detection model with high priority
CREATE MODEL fraud_detector
INPUT (
  user_id BIGINT,
  amount DOUBLE,
  merchant_id STRING,
  device_fingerprint STRING
)
OUTPUT (fraud_score FLOAT)
WITH (
  'provider' = 'triton',
  'endpoint' = 'http://triton-server:8000/v2/models',
  'model-name' = 'fraud_detection_model',
  'timeout' = '5000',
  'priority' = '200'  -- High priority for critical transactions
);

CREATE TEMPORARY TABLE transactions (
  transaction_id STRING,
  user_id BIGINT,
  amount DECIMAL(10, 2),
  merchant_id STRING,
  device_fingerprint STRING,
  transaction_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'transactions',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

-- Flag suspicious transactions
SELECT
  transaction_id,
  user_id,
  amount,
  fraud_score,
  CASE
    WHEN fraud_score > 0.8 THEN 'HIGH_RISK'
    WHEN fraud_score > 0.5 THEN 'MEDIUM_RISK'
    ELSE 'LOW_RISK'
  END AS risk_level
FROM ML_PREDICT(
  TABLE transactions,
  MODEL fraud_detector,
  DESCRIPTOR(user_id, amount, merchant_id, device_fingerprint)
)
WHERE fraud_score > 0.5;  -- Alert on suspicious transactions

예시 4: 추천 시스템 (Example 4: Recommendation System)

사용자 행동에 기반한 제품 추천:

-- Register recommendation model
CREATE MODEL recommender
INPUT (
  user_features ARRAY<FLOAT>,
  browsing_history ARRAY<STRING>,
  context_features ARRAY<FLOAT>
)
OUTPUT (recommended_products ARRAY<STRING>, scores ARRAY<FLOAT>)
WITH (
  'provider' = 'triton',
  'endpoint' = 'http://triton-server:8000/v2/models',
  'model-name' = 'product_recommender',
  'model-version' = '2'
);

CREATE TEMPORARY TABLE user_activity (
  user_id BIGINT,
  user_features ARRAY<FLOAT>,
  browsing_history ARRAY<STRING>,
  context_features ARRAY<FLOAT>,
  event_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_events',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

-- Generate personalized recommendations
SELECT
  user_id,
  recommended_products,
  scores,
  event_time
FROM ML_PREDICT(
  TABLE user_activity,
  MODEL recommender,
  DESCRIPTOR(user_features, browsing_history, context_features)
);

예시 5: 개체명 인식 (NER) (Example 5: Named Entity Recognition (NER))

텍스트 문서에서 엔티티를 추출합니다:

-- Register NER model with compression for large documents
CREATE MODEL ner_model
INPUT (document_text STRING)
OUTPUT (entities ARRAY<STRING>, entity_types ARRAY<STRING>)
WITH (
  'provider' = 'triton',
  'endpoint' = 'http://triton-server:8000/v2/models',
  'model-name' = 'bert_ner',
  'compression' = 'gzip'
);

CREATE TEMPORARY TABLE documents (
  doc_id STRING,
  document_text STRING,
  source STRING,
  created_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'documents',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

-- Extract named entities
SELECT
  doc_id,
  entities,
  entity_types,
  source
FROM ML_PREDICT(
  TABLE documents,
  MODEL ner_model,
  DESCRIPTOR(document_text)
);

예시 6: 상태 저장 시퀀스 모델 (Example 6: Stateful Sequence Model)

시퀀스 추적과 함께 상태 저장 모델(RNN/LSTM)을 사용합니다:

-- Register stateful conversation model
CREATE MODEL conversation_model
INPUT (message_text STRING)
OUTPUT (bot_response STRING)
WITH (
  'provider' = 'triton',
  'endpoint' = 'http://triton-server:8000/v2/models',
  'model-name' = 'chatbot_lstm',
  'sequence-id' = 'conv-001',  -- Unique sequence ID
  'sequence-start' = 'true',
  'sequence-end' = 'false'
);

CREATE TEMPORARY TABLE chat_messages (
  message_id STRING,
  user_id BIGINT,
  message_text STRING,
  timestamp TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'chat',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

-- Process conversation with context
SELECT
  message_id,
  user_id,
  bot_response,
  timestamp
FROM ML_PREDICT(
  TABLE chat_messages,
  MODEL conversation_model,
  DESCRIPTOR(message_text)
);

예시 7: 배치 추론 (Example 7: Batch Inference)

과거 데이터에 배치 추론을 수행합니다:

-- Register model for batch processing
CREATE MODEL batch_classifier
INPUT (features ARRAY<DOUBLE>)
OUTPUT (prediction STRING, confidence DOUBLE)
WITH (
  'provider' = 'triton',
  'endpoint' = 'http://triton-server:8000/v2/models',
  'model-name' = 'classifier',
  'timeout' = '60000'
);

-- Batch source table
CREATE TEMPORARY TABLE historical_data (
  id BIGINT,
  features ARRAY<DOUBLE>
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs:///data/historical',
  'format' = 'parquet'
);

-- Batch inference with results written to sink
CREATE TEMPORARY TABLE classification_results (
  id BIGINT,
  prediction STRING,
  confidence DOUBLE
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs:///results/classifications',
  'format' = 'parquet'
);

INSERT INTO classification_results
SELECT id, prediction, confidence
FROM ML_PREDICT(
  TABLE historical_data,
  MODEL batch_classifier,
  DESCRIPTOR(features)
);

예시 8: 보안된 Triton 서버 (Example 8: Secured Triton Server)

인증으로 보안된 Triton 서버에 접근합니다:

-- Register model with authentication
CREATE MODEL secure_model
INPUT (data STRING)
OUTPUT (result STRING)
WITH (
  'provider' = 'triton',
  'endpoint' = 'https://secure-triton:8000/v2/models',
  'model-name' = 'private_model',
  'auth-token' = 'Bearer your-token-here',
  'custom-headers' = '{"X-API-Key": "your-api-key", "X-Client-ID": "flink-job-123"}'
);

SELECT id, result
FROM ML_PREDICT(
  TABLE sensitive_data,
  MODEL secure_model,
  DESCRIPTOR(data)
);

SQL에 민감한 토큰을 하드코딩하지 마세요. Flink의 비밀 관리 또는 환경 변수를 사용하세요.

예시 9: 배치 차원을 평탄화한 배열 타입 (Example 9: Array Type with Flatten Batch Dimension)

배치 차원 없이 배열 입력을 받는 모델의 경우:

-- Create model with array input and flatten batch dimension
CREATE MODEL triton_vector_model
INPUT (input_vector ARRAY<FLOAT>)
OUTPUT (output_vector ARRAY<FLOAT>)
WITH (
    'provider' = 'triton',
    'endpoint' = 'http://localhost:8000/v2/models',
    'model-name' = 'vector-transform',
    'model-version' = '1',
    'flatten-batch-dim' = 'true'  -- Flatten [1,N] to [N]
);

-- Use the model for inference
CREATE TEMPORARY TABLE vector_input (
    id BIGINT,
    features ARRAY<FLOAT>
) WITH (
    'connector' = 'datagen',
    'fields.features.length' = '128'  -- 128-dimensional vector
);

SELECT id, output_vector
FROM ML_PREDICT(
  TABLE vector_input,
  MODEL triton_vector_model,
  DESCRIPTOR(features)
);

예시 10: 고급 구성 (Example 10: Advanced Configuration)

종합 설정이 필요한 프로덕션 환경의 경우:

CREATE MODEL triton_advanced_model
INPUT (`input` STRING)
OUTPUT (`output` STRING)
WITH (
    'provider' = 'triton',
    'endpoint' = 'https://triton.example.com/v2/models',
    'model-name' = 'advanced-nlp-model',
    'model-version' = 'latest',
    'timeout' = '15000',
    'priority' = '100',
    'auth-token' = 'Bearer your-auth-token-here',
    'custom-headers' = '{"X-Custom-Header": "custom-value", "X-Request-ID": "req-123"}',
    'compression' = 'gzip'
);

모델 옵션 (Model Options)

필수 옵션 (Required Options)

키	기본값	타입	설명
endpoint	(none)	String	Triton Inference Server 엔드포인트의 전체 URL, 예: https://triton-server:8000/v2/models. HTTP와 HTTPS 모두 지원하며, 프로덕션에는 HTTPS를 권장합니다.
model-name	(none)	String	Triton 서버에서 호출할 모델 이름.
model-version	"latest"	String	사용할 모델 버전. 기본값은 'latest'.
timeout	30 s	Duration	HTTP 요청 시간 초과(connect + read + write). 개별 요청별로 적용되며 Flink의 비동기 시간 초과와는 별개입니다. 기본값 30초.

선택 옵션 (Optional Options)

키	기본값	타입	설명
auth-token	(none)	String	보안된 Triton 서버를 위한 인증 토큰.
compression	(none)	String	요청 본문의 압축 알고리즘. 현재 gzip만 지원됩니다. 활성화하면 요청 본문이 압축되어 네트워크 대역폭을 줄입니다.
custom-headers	(none)	Map	키-값 쌍으로 된 커스텀 HTTP 헤더. 예: 'X-Custom-Header:value,X-Another:value2'
flatten-batch-dim	false	Boolean	배열 입력에 대해 배치 차원을 평탄화할지 여부. true면 shape [1,N]이 [N]이 됩니다. 기본값 false.
priority	(none)	Integer	요청 우선순위 수준(0-255). 값이 높을수록 우선순위가 높습니다.
sequence-end	false	Boolean	이 요청이 상태 저장 모델의 시퀀스 끝을 표시하는지 여부. true면 Triton은 이 요청 처리 후 모델 상태를 해제합니다. 자세한 내용은 Triton Stateful Models를 참고하세요.
sequence-id	(none)	String	상태 저장 모델을 위한 시퀀스 ID. 시퀀스는 요청 간 상태를 유지하기 위해 같은 모델 인스턴스로 라우팅되어야 하는 일련의 추론 요청을 나타냅니다(예: RNN/LSTM 모델). 자세한 내용은 Triton Stateful Models를 참고하세요.
sequence-start	false	Boolean	이 요청이 상태 저장 모델의 새 시퀀스 시작을 표시하는지 여부. true면 Triton은 이 요청 처리 전에 모델 상태를 초기화합니다. 자세한 내용은 Triton Stateful Models를 참고하세요.

스키마 요구 사항 (Schema Requirement)

입력 타입	출력 타입	설명
BOOLEAN, TINYINT, SMALLINT, INT, BIGINT	BOOLEAN, TINYINT, SMALLINT, INT, BIGINT	정수 타입 추론
FLOAT, DOUBLE	FLOAT, DOUBLE	부동소수점 타입 추론
STRING	STRING	텍스트-텍스트 추론(분류, 생성 등)
ARRAY<숫자 타입>	ARRAY<숫자 타입>	배열 추론(벡터, 텐서 등). 숫자 타입 배열을 지원합니다.

참고: 입력과 출력 타입은 Triton 모델 구성에 정의된 타입과 일치해야 합니다. 모델의 예상 입력/출력 타입을 확인하려면 Triton 서버를 쿼리하세요:

curl http://triton-server:8000/v2/models/{model_name}/config

Triton 서버 설정 (Triton Server Setup)

이 통합을 사용하려면 실행 중인 Triton Inference Server가 필요합니다. 기본 설정 가이드는 다음과 같습니다:

Docker 사용 (Using Docker)

# Pull Triton server image
docker pull nvcr.io/nvidia/tritonserver:23.10-py3

# Run Triton server with your model repository
docker run --rm -p 8000:8000 -p 8001:8001 -p 8002:8002 \
  -v /path/to/your/model/repository:/models \
  nvcr.io/nvidia/tritonserver:23.10-py3 \
  tritonserver --model-repository=/models

모델 저장소 구조 (Model Repository Structure)

모델 저장소는 이 구조를 따라야 합니다:

model_repository/
├── text-classification/
│   ├── config.pbtxt
│   └── 1/
│       └── model.py  # or model.onnx, model.plan, etc.
└── other-model/
    ├── config.pbtxt
    └── 1/
        └── model.savedmodel/

예시 모델 구성 (Example Model Configuration)

텍스트 분류 모델을 위한 config.pbtxt 예시입니다:

name: "text-classification"
platform: "python"
max_batch_size: 8
input [
  {
    name: "INPUT_TEXT"
    data_type: TYPE_STRING
    dims: [ 1 ]
  }
]
output [
  {
    name: "OUTPUT_TEXT"
    data_type: TYPE_STRING
    dims: [ 1 ]
  }
]

성능 고려 사항 (Performance Considerations)

1. 연결 풀링 (Connection Pooling): HTTP 클라이언트는 효율성을 위해 풀링되고 재사용됩니다.
2. 비동기 처리 (Asynchronous Processing): 비블로킹 요청이 스레드 부족(thread starvation)을 방지합니다.
3. 배치 처리 (Batch Processing): 최적의 처리량을 위해 배치 크기를 구성합니다:
   • 단순 모델: 배치-크기 1-4
   • 중간 모델: 배치-크기 4-16
   • 복잡한 모델: 배치-크기 16-32
4. 리소스 관리 (Resource Management): HTTP 리소스의 자동 정리.
5. 시간 초과 구성 (Timeout Configuration): 모델 복잡성에 따라 적절한 시간 초과 값을 설정합니다:
   • 단순 모델: 1-5초
   • 중간 모델(예: BERT): 5-30초
   • 복잡한 모델(예: GPT): 30-120초
6. 재시도 전략 (Retry Strategy): 일시적 실패를 처리하기 위해 재시도 횟수를 구성합니다.
7. 압축 (Compression): 1KB 초과 페이로드에 gzip 압축을 활성화합니다.
8. 병렬도 (Parallelism): Flink 병렬도를 Triton 서버 용량에 맞춥니다.

모범 사례 (Best Practices)

모델 버전 관리 (Model Version Management)

일관성을 보장하기 위해 프로덕션에서 모델 버전을 고정합니다:

'model-version' = '3'  -- Pin to version 3 instead of 'latest'

오류 처리 (Error Handling)

실패 시 기본값을 사용합니다:

SELECT COALESCE(output, 'UNKNOWN') AS prediction
FROM ML_PREDICT(...)

리소스 구성 (Resource Configuration)

고처리량 시나리오를 위해 충분한 메모리와 네트워크 버퍼를 구성합니다:

taskmanager.memory.managed.size: 2gb
taskmanager.network.memory.fraction: 0.2

오류 처리 (Error Handling)

이 통합은 포괄적인 오류 처리를 포함합니다:

• 연결 오류 (Connection Errors): 지수 백오프를 사용한 자동 재시도
• 시간 초과 처리 (Timeout Handling): 구성 가능한 요청 시간 초과
• HTTP 오류 (HTTP Errors): Triton 서버의 상세 오류 메시지
• 직렬화 오류 (Serialization Errors): JSON 파싱 및 검증 오류

모니터링과 디버깅 (Monitoring and Debugging)

통합을 모니터링하려면 디버그 로깅을 활성화합니다:

# In log4j2.properties
logger.triton.name = org.apache.flink.model.triton
logger.triton.level = DEBUG

이것은 다음에 대한 상세 로그를 제공합니다:

• HTTP 요청/응답 세부 정보
• 클라이언트 연결 관리
• 오류 조건과 재시도
• 성능 메트릭

문제 해결 (Troubleshooting)

연결 문제 (Connection Issues)

• Triton 서버가 실행 중인지 확인: curl http://triton-server:8000/v2/health/ready
• 네트워크 연결과 방화벽 규칙 확인
• 엔드포인트 URL이 올바른 프로토콜(http/https)을 포함하는지 확인

시간 초과 오류 (Timeout Errors)

• 시간 초과 값 증가: 'timeout' = '60000'
• Triton 서버 리소스 사용량(CPU/GPU) 확인
• 느린 모델 실행을 위해 Triton 서버 로그 모니터링

타입 불일치 (Type Mismatch)

• 모델 스키마 확인: curl http://triton-server:8000/v2/models/{model}/config
• Flink 타입을 명시적으로 캐스트: CAST(value AS FLOAT)
• 배열 차원이 모델 기대에 맞는지 확인

높은 지연 (High Latency)

• 요청 압축 활성화: 'compression' = 'gzip'
• Triton 서버 인스턴스 증가
• Triton 서버 구성에서 동적 배칭 사용
• Flink와 Triton 간 네트워크 지연 확인

의존성 (Dependencies)

Triton 모델 함수를 사용하려면 Flink 애플리케이션에 다음 의존성을 포함해야 합니다:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-model-triton</artifactId>
    <version>${flink.version}</version>
</dependency>

추가 정보 (Further Information)

• Triton Inference Server Documentation
• Triton Model Configuration
• Flink Async I/O
• Flink Metrics

더 알아보기 (Learn more)

• AI Model Connectors
• OpenAI Model Function
• ML_PREDICT