프로그래매틱 광고(Programmatic advertising)에서는 매 밀리초(millisecond)가 중요합니다. 실시간 입찰 (RTB, Real-time bidding) 시스템은 인상(impression)을 낙찰받기 위해 경매에서 경쟁하며 100ms 미만의 시간 안에 입찰가를 반환해야 하는 동시에, 오디언스 세분화(audience segmentation), 개인화(personalization), 모델 재학습(model retraining)을 지원하기 위해 조회(views) 및 디바이스/캠페인 메타데이터와 같은 클릭스트림(clickstream) 이벤트를 동시에 캡처해야 합니다. 저는 지연 시간 급증(latency spikes) 없이 이 두 가지를 대규모로 수행하는 방법을 알아보고 싶었습니다.

이를 위해 저는 Redpanda Connect를 통해 클릭스트림 이벤트를 수집하고, Redpanda Iceberg Topics를 사용하여 Apache Iceberg 테이블 형태로 데이터 레이크(data lake)에 기록하며, 이를 소규모 AI 스코어링 모델(AI scoring model)을 학습시키고 서빙(serve)하는 데 사용하여 예측값을 다시 입찰 루프(bidder loop)로 전달하는 RTB 스타일의 파이프라인을 구축했습니다.

제 목표는 OpenRTB를 엔드 투 엔드(end-to-end)로 완전히 구현하는 것이 아니라, 실제 운영 환경의 RTB 아키텍처로 확장 가능한 고처리량(high-throughput), 저지연(low-latency) 파이프라인의 뼈대를 구축하는 것이었습니다. 여러분도 따라 하며 직접 구축해 볼 수 있도록 제가 어떻게 구성했는지 소개하겠습니다.

RTB 입찰 최적화

세 가지 반복적인 문제를 가진 RTB 파이프라인을 상상해 보세요:

• 트래픽 급증 시 발생하는 지연 시간 급증 (응답 시간 범위를 초과하면 경매에서 탈락합니다).
• 피크 부하(peak load)를 견디기 위해 과도하게 프로비저닝(overprovisioning)함으로써 발생하는 높은 인프라 비용.
• 과거 데이터가 사일로(silos)에 존재하여 쿼리 및 재학습이 어렵기 때문에 발생하는 느린 모델 갱신.

목표는 시스템을 개선하여 다음과 같은 기능을 수행하도록 하는 것입니다:

• 클릭스트림 이벤트(조회/클릭/디바이스/캠페인)를 실시간으로 수집.
• 분석 및 모델 학습을 위해 이벤트를 Iceberg 테이블로 비용 효율적으로 저장.
• AI를 사용하여 인상 가치(CTR proxy/value score)를 예측.
• 입찰자(bidder)에게 예측값을 스트리밍하여 더 빠르고 스마트한 결정을 내릴 수 있도록 지원.

대략적인 아키텍처는 다음과 같습니다:

시스템을 개편하기 위해 다음 구성 요소들을 구현하게 됩니다:

• Redpanda Connect는 입찰 요청(bid requests)의 HTTP 인제스션(ingestion)을 처리하며, 요청을 소비(consume)하고 예측값을 생성하는 "접착제(glue)" 역할을 합니다.
• Redpanda Iceberg Topics는 데이터를 오브젝트 스토리지(object storage) 내의 Iceberg 테이블로 스트리밍합니다. Redpanda는 데이터를 Iceberg 호환 형식으로 기록하여, Iceberg 클라이언트로 쿼리할 수 있게 합니다.
• Spark/Jupyter (실습용)는 Iceberg 테이블을 쿼리하여 학습을 위한 데이터셋을 생성합니다.
• 소규모 추론(inference) 서비스는 입찰 요청의 점수를 매기고 bid_predictions를 발행(publish)합니다.
• 간단한 입찰자(bidder) 루프는 예측값을 소비하여 얼마나 공격적으로 입찰할지 결정합니다.

사전 요구 사항

이 튜토리얼을 간단하고 반복 가능하게 유지하기 위해, Redpanda의 공식 Iceberg Docker Compose 실습(lab)을 기본 환경으로 사용합니다. 여기에는 Redpanda, MinIO (S3 호환 스토리지), 그리고 Iceberg 테이블을 쿼리하도록 설정된 Spark + Jupyter가 포함되어 있습니다.

또한 다음이 필요합니다:

• Docker 및 Docker Compose
• rpk CLI
• Python 3.11 이상 (소규모 추론 서비스 및 생성기용)

Iceberg 플레이그라운드 실행하기 (Redpanda, MinIO, Spark/Jupyter)

먼저, 실습 리포지토리를 클론(clone)하고 프로젝트 디렉토리로 이동합니다:

git clone https://github.com/redpanda-data/redpanda-labs.git && cd redpanda-labs/docker-compose/iceberg

이 실습은 Redpanda >= 24.3.1 버전을 요구합니다. 필요한 Redpanda 버전을 정의합니다:

export REDPANDA_VERSION=v25.3.2
export REDPANDA_CONSOLE_VERSION=v3.3.2

환경을 시작합니다. 이를 통해 데이터를 생성하는 것부터 Spark/Jupyter에서 Iceberg 테이블을 쿼리하는 것까지의 전체 파이프라인(pipeline)이 제공됩니다:

docker compose build && docker compose up

다음으로, 실습을 위한 rpk 프로필을 생성합니다:

rpk profile create docker-compose-iceberg \
  --set=admin_api.addresses=localhost:19644 \
  --set=brokers=localhost:19092 \
...

RTB 토픽 생성하기

이 섹션에서는 세 가지 토픽을 생성합니다:

• clickstream (Iceberg 활성화)
• bid_requests (점수를 매길 메시지)
• bid_predictions (점수가 매겨진 출력값)

clickstream의 경우, key_value 모드로 시작하십시오. 이는 HTTP 인제스션 (Ingestion)을 위한 가장 쉬운 방법이기 때문입니다 (JSON을 POST하고 이를 "value"로 취급할 수 있습니다).

다음 명령어 세트를 실행하여 토픽을 생성합니다:

rpk topic create clickstream --topic-config=redpanda.iceberg.mode=key_value ;
rpk topic create bid_requests ;
rpk topic create bid_predictions ;

Iceberg 토픽은 redpanda.iceberg.mode를 통해 제어됩니다. Iceberg가 활성화된 토픽으로 데이터를 프로듀스 (Produce)하면, 해당 데이터는 Iceberg 클라이언트가 소비 (Consume)할 수 있도록 오브젝트 스토리지 (Object Storage)에서 사용할 수 있게 됩니다.

Redpanda Connect를 사용한 clickstream 인제스션 실행

AdTech에서 클릭스트림 (Clickstream)은 종종 HTTP 컬렉터 (Collectors)를 통해 전달됩니다. Redpanda Connect에는 POST된 이벤트를 수신하기 위한 http_server 입력 (Input)이 있습니다.

로컬 컴퓨터에서 작업 디렉토리를 선택하고, connect-clickstream.yaml이라는 YAML 파일을 생성한 다음, 다음 내용을 붙여넣으십시오:

input:
  http_server:
    address: 0.0.0.0:4196
...

kafka_franz는 Redpanda Connect의 Kafka 출력 (Output)입니다.

참고: Redpanda Cloud 문서에서 kafka_franz는 통합된 Redpanda 컴포넌트를 위해 사용 중단 (Deprecated)된 것으로 표시되어 있습니다. 대상 환경이 Cloud인 경우, 그에 따라 출력을 전환하는 것을 고려하십시오.

실습과 동일한 Docker 네트워크 상에서 컨테이너로 Redpanda Connect를 실행하십시오 (Docker Compose 프로젝트 이름에 따라 네트워크 이름을 조정해야 할 수도 있습니다):

docker run --rm -it \
  --network redpanda-labs_default \
  -p 4196:4196 \
...

그런 다음, 몇 개의 샘플 클릭스트림 이벤트를 전송합니다:

curl -X POST http://localhost:4196/events \
  -H 'Content-Type: application/json' \
  -d '{"user_id":101,"ad_id":55,"campaign_id":9,"event_type":"click","clicked":1,"ts":"2025-12-17T09:00:00Z"}'
...

생성된 클릭스트림 (clickstream) 데이터를 검증하려면, 새 터미널 창에서 다음 명령어를 사용하여 clickstream 데이터를 소비 (consume) 하세요:

rpk topic consume clickstream

출력 결과는 다음과 같아야 합니다:

{
  "topic": "clickstream",
  "value": "{\"user_id\":101,\"ad_id\":55,\"campaign_id\":9,\"event_type\":\"click\",\"clicked\":1,\"ts\":\"2025-12-17T09:00:00Z\"}",
...

Iceberg 쓰기 검증

브라우저에서 http://localhost:9001/browser를 열고 자격 증명 minio / minio123을 입력하세요:

다른 탭을 열고 Jupyter Labs 페이지인 http://localhost:8888로 이동하세요.

랩 노트북 (lab notebook)이 Redpanda 토픽 (topic)으로부터 생성된 Iceberg 테이블을 쿼리하는 과정을 안내할 것입니다.

Spark SQL을 사용하여 간단한 무결성 검사 (sanity check)를 수행할 수도 있습니다. 다음 명령어를 실행하여 spark-iceberg와 spark-sql을 시작하세요:

docker exec -it spark-iceberg spark-sql

열린 인터페이스에서 Iceberg 테이블을 쿼리할 수 있습니다.

참고: 정확한 카탈로그/테이블 명명 규칙은 랩 노트북에 표시되어 있습니다. 이 랩에서는 lab.redpanda.<topic>을 쿼리하는 것을 시연합니다.

Iceberg 데이터로부터 "입찰가 (bid value)" 모델 구축하기

작은 "입찰가 (bid value)" 모델을 만들기 위해서는, 먼저 데이터를 JSON에서 파싱 (parsing) 하여 준비한 다음, Jupyter Labs 컨테이너의 디스크에 parquet 파일로 저장해야 합니다. 그 후 저장된 학습 데이터셋을 사용하여 모델을 훈련할 수 있습니다.

데이터 준비

이 시점에서 여러분은 기본 경로인 HTTP -> Redpanda -> Iceberg를 검증했습니다. 다음 과제는 신선한 행동 데이터 (behavioral data)를 기반으로 모델을 훈련하고, 이를 저지연 (low-latency) 스코어링 서비스로 전환하는 것입니다.

Iceberg 토픽을 key_value 모드로 생성했기 때문에, Iceberg 테이블은 페이로드 (payload)를 바이너리 value 컬럼에 저장합니다 (레코드 메타데이터를 보유하는 redpanda 구조체 포함).

이는 훈련 워크플로가 두 단계로 나뉜다는 것을 의미합니다:

• 추출 (Extract): value를 캐스팅/디코딩하고 JSON을 타입이 지정된 컬럼 (typed columns)으로 파싱
• 훈련 (Train): 경량 모델을 적합 (fit)시키고 API에서 로드할 수 있는 아티팩트 (artifact)를 저장

브라우저의 Jupyter로 돌아가 새로운 Python 노트북을 생성하세요. 첫 번째 셀에서 Iceberg 테이블을 로드합니다. 이 실습에서 테이블은 lab.redpanda.<topic> 형식으로 쿼리할 수 있습니다 (실습에서 lab.redpanda.value_schema_id_prefix를 쿼리하는 것을 명시적으로 보여주므로, 여러분의 clickstream 토픽은 lab.redpanda.clickstream이 될 것입니다).

첫 번째 줄에 다음 스크립트를 입력하고 실행하세요:

clickstream_raw = spark.table("lab.redpanda.clickstream")
clickstream_raw.printSchema()

다음 항목을 포함하는 스키마가 보여야 합니다:

• redpanda 구조체 (struct) 컬럼 (메타데이터)
• value 컬럼 (바이너리 페이로드)

원시 페이로드를 확인하려면, 노트북의 두 번째 줄에 다음 스크립트 내용을 복사하여 붙여넣으세요:

from pyspark.sql.functions import col

(clickstream_raw
...

이벤트가 JSON이었다면 json_value에서 JSON 문자열을 볼 수 있어야 합니다. 만약 null이 보이거나 에러가 발생한다면, 보통 다음 두 가지 중 하나를 의미합니다:

• 실제로 JSON을 발행(publish)하지 않았음 (curl을 사용한 경우라면 가능성이 낮음).
• 바이트가 UTF-8이 아니어서 캐스팅 (cast) 시 깔끔하게 디코딩되지 않음.

이제 JSON 페이로드를 타입이 지정된 컬럼으로 붙여넣어야 합니다. 이를 위해 클릭스트림 수집기 (clickstream collector)로부터 기대되는 스키마를 정의해야 합니다. Jupyter 노트북에 다음 내용을 추가하고 실행하세요:

from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

...

값을 파싱하려면 다음 스크립트를 추가하고 노트북에서 실행하세요:

parsed = (clickstream_raw
    .select(from_json(col("value").cast("string"), payload_schema).alias("e"))
    .select("e.*")
...

다음과 같이 정제된 컬럼들을 확인할 수 있어야 합니다:

• user_id, ad_id, campaign_id (정수형 (Integer))
• event_type (문자열 (String))
• clicked (0 또는 1)

만약 clicked 이벤트가 가끔 누락되는 경우, (데모를 위해) 기본값을 0으로 설정할 수 있습니다:

from pyspark.sql.functions import when, lit

parsed = parsed.withColumn("clicked", when(col("clicked").isNull(), lit(0)).otherwise(col("clicked")))

이 튜토리얼에서는 수백만 개의 행을 대상으로 직접 학습하지 않습니다. 노트북의 속도를 유지하고 반복 가능하게 만들기 위해 샘플을 추출해야 합니다:

training_df = parsed.limit(100_000)

training_path = "/home/jovyan/work/training_clickstream.parquet"
...

이 시점에서, Jupyter 컨테이너 내부 디스크에 안정적인 데이터셋이 준비되었습니다.

모델 학습 (Training the Model)

모델을 학습하기 위해, event_type에 대한 원-핫 인코딩 (One-hot encoding)을 포함한 로지스틱 회귀 (Logistic regression) 파이프라인을 사용합니다. 이는 의도적으로 작게 구성된 "데모급" 모델이지만, 파이프라인을 증명하기에는 충분합니다.

동일한 Jupyter 노트북의 다음 빈 셀에 아래 내용을 복사하여 붙여넣으세요:

# 노트북 환경에서 필요한 경우:
# %pip install -q scikit-learn joblib pandas pyarrow

...

그 다음 아티팩트 (Artifact)를 저장합니다:

artifact_path = "/home/jovyan/work/bid_value_model.joblib"
joblib.dump(model, artifact_path)
artifact_path

이제 소규모 추론 서비스 (Inference service)로 쉽게 전달할 수 있는 모델 아티팩트를 확보했습니다.

모델을 추론 엔드포인트로 서빙하기 (Serving the model as an inference endpoint)

학습된 모델 아티팩트(bid_value_model.joblib)가 준비되었으므로, 이를 온라인 RTB 경로에서 사용할 수 있도록 만들어야 합니다. 이를 위해 모델 아티팩트를 Jupyter 컨테이너 외부로 복사한 다음, 미리 생성된 추론 서비스를 실행합니다.

Jupyter 컨테이너 외부로 모델 아티팩트 복사하기

Jupyter/Spark 컨테이너 내부의 /home/jovyan/work/bid_value_model.joblib 경로에 아티팩트 (artifact)를 저장했을 것입니다.

호스트 머신(host machine)에서 추론 서비스 (inference service) 파일들을 보관할 디렉토리에 터미널을 열고, 모델을 외부로 복사합니다: