Kafka 연결, 스트리밍 수집과 내보내기

Kafka 커넥터는 Kafka 토픽의 메시지를 Foundry 스트림으로 수집하고 Foundry 스트림을 외부 Kafka 토픽으로 내보냅니다. 원천 데이터 탐색, 스트리밍 동기화와 스트리밍 내보내기는 일반 제공 기능입니다.

메시지와 처리량

기본 수집 결과에는 바이너리 key와 value 열이 있습니다. 메시지 내용은 해석하지 않고 value에 저장하며, 키가 없는 메시지는 key가 null입니다. 예를 들어 키가 London인 메시지의 값이 {"firstName":"John","lastName":"Doe"}여도 기본 수집 단계에서는 이 값을 객체 필드로 나누지 않습니다. JSON 해석은 후속 스트리밍 변환에서 수행합니다. Schema Registry를 연결한 경우에는 아래의 스키마 해석 경로를 사용합니다.

기본 소비자 스레드는 하나입니다. 스레드를 늘리면 브로커가 각 소비자에게 토픽 파티션의 일부를 할당합니다. 스레드 수가 파티션 수보다 많으면 남는 스레드는 유휴 상태가 되므로 처리량이 늘지 않습니다.

스트리밍 동기화는 장시간 실행하는 작업이지만 재시작에 따른 중단이 발생합니다.

소스 생성과 네트워크

1. Data Connection에서 + New Source를 선택하고 Kafka 소스를 선택합니다. 목록에 없으면 Palantir 지원팀에 접근 활성화를 요청합니다.
2. 실행 위치와 네트워크 경로를 정합니다. Foundry 워커에서 인터넷으로 직접 연결하거나 내부망에 도달하는 에이전트 프록시 경로를 구성합니다. Legacy 에이전트 워커는 기존 구성의 제약을 확인하고 사용합니다.
3. Bootstrap servers에 HOST:PORT 형식의 브로커를 한 줄에 하나씩 입력합니다. 필수 항목이며 기본 서버는 없습니다.
4. 인증과 인증서를 설정하고 연결 구성을 저장합니다. 직접 연결에서는 모든 부트스트랩 서버 호스트에 DNS 외부 통신 정책을 설정해야 합니다. 실제 Kafka 브로커 호스트에도 도달할 수 있어야 합니다.

Kafka 자격 증명에는 Topic 리소스 권한이 필요합니다. 탐색과 스트리밍 수집에는 Read, 스트리밍 내보내기에는 Write를 부여합니다.

인증 설정

인증 방식에는 SSL, 사용자명과 비밀번호, OAuth 2.0, Azure AD, Kerberos와 NONE이 있습니다. 아래 항목의 기본값 없음은 값이 자동으로 제공되지 않는다는 뜻입니다.

SSL과 SASL

SSL 인증은 Kafka의 SSL 또는 SASL_SSL 프로토콜에 대응합니다. SASL은 인증 메커니즘을 선택하는 기능이며 TLS가 통신을 암호화합니다.

SASL을 활성화하면 다음 항목을 구성합니다.

OAuth 2.0

Client Credentials Grant만 지원합니다.

Azure Event Hubs

Azure AD 모드는 Azure Event Hubs의 Kafka 인터페이스에 적용합니다. Event Hubs 접근 권한을 가진 서비스 주체를 만들고 Tenant ID, Client ID, Client secret을 모두 입력합니다. 세 항목에는 기본값이 없습니다.

SAS 토큰 방식으로 연결할 때에는 사용자명과 비밀번호 인증을 선택합니다. 사용자명은 따옴표 없는 $ConnectionString, 비밀번호는 Event Hubs 연결 문자열입니다.

인증 없음

NONE은 Kafka의 PLAINTEXT에 대응합니다. 인증이나 SSL 없이 데이터를 전송하므로 암호화되지 않은 메시지가 네트워크를 통과합니다. 안전하게 통제한 네트워크 내부에서만 사용합니다.

인증서와 mTLS

SSL 연결은 서버 인증서를 검증합니다. Foundry 워커와 에이전트 워커는 기본적으로 일반적인 인증서 체인을 신뢰합니다. 자체 서명 인증서나 TLS 검사 장비 때문에 추가 신뢰 설정이 필요하면 Kafka 관리자가 제공한 올바른 인증서를 사용합니다. 호스트명 검사를 끄는 것과 신뢰할 인증서를 추가하는 것은 서로 다른 설정입니다.

Foundry 워커는 Control Panel의 외부 통신 인증서 관리에서 인증서를 중앙 관리하고 소스에 적용할 수 있습니다. mTLS에서는 서버 인증서 외에 클라이언트 인증서와 개인 키도 준비합니다.

직접 연결의 경우 소스 설정에서 Configure client certificates and private key를 열고 별칭을 kafka로 지정한 뒤 개인 키와 클라이언트 인증서를 추가합니다.

Legacy 에이전트 워커에서는 기본 bootstrapper 키 저장소를 직접 수정하면 재시작할 때 변경이 사라집니다. 에이전트 실행 사용자로 기본 keyStore.jks를 별도 경로에 복사하고 JDK의 keytool로 고객 키를 가져옵니다. 일반적인 에이전트 키 저장소 구성에서는 가져온 개인 키의 비밀번호와 키 저장소 비밀번호를 같게 맞춥니다. Kafka의 사용자 정의 키 비밀번호 설정을 쓰는 경우에는 앞의 조건을 따릅니다.

Data Connection의 해당 에이전트에서 Agent settings, Manage Configuration, Advanced를 차례로 엽니다. Agent 탭의 security.keyStore와 security.keyStorePassword를 별도 저장소에 맞게 설정하고, Explorer 탭에서는 security.keyStorePath와 security.keyStorePassword를 설정합니다. 두 탭의 경로 필드명이 다릅니다. 기본 키 저장소 비밀번호는 keystore이며 실제 저장소에 설정한 값을 사용해야 합니다. 저장한 뒤 에이전트를 재시작합니다. Bootstrapper 구성은 바꾸지 않습니다.

수집과 Schema Registry

소스에서 스트리밍 동기화를 만들고 수집할 토픽과 출력 스트림을 지정합니다. 처리량과 실행 설정은 스트리밍 동기화 설정을 따릅니다.

Schema Registry는 토픽 스키마의 버전 이력을 관리합니다. Avro, Protobuf와 JSON 스키마를 지원하며 SerDes, 즉 직렬화기와 역직렬화기가 스키마와 바이너리 메시지 사이를 변환합니다. 대상 토픽의 스키마를 레지스트리에 등록하고 소스 구성의 Schema Registry URL에 주소를 추가하면 커넥터가 메시지의 바이트를 대응하는 Foundry 자료형으로 변환합니다.

스트리밍 내보내기

소스에서 내보내기 기능을 활성화하고 새 스트리밍 내보내기를 생성합니다. 입력 Foundry 스트림과 목적지 Kafka 토픽을 연결합니다.

기존 내보내기 작업 유지보수

새 구성에는 위의 내보내기 기능을 사용하고 기존 export task도 이 방식으로 이관합니다. 이전 작업 구성은 소스가 있는 프로젝트 폴더에서 커넥터를 우클릭한 뒤 Create Data Connection Task로 생성하던 방식입니다. 해당 작업을 유지보수할 때에는 왼쪽 패널의 Source를 확인하고 다음 입출력을 구분합니다.

• dataset이라는 Streaming dataset 입력은 외부로 보낼 Foundry 데이터입니다.
• dataset이라는 Streaming export 출력은 작업 실행, 예약과 모니터링에 쓰는 리소스입니다.

batchOptions.maxMilliseconds와 batchOptions.maxRecords는 모두 필수이며 기본값이 없습니다. 대기 시간과 메시지 개수 중 먼저 충족하는 조건에서 배치를 보냅니다. 값을 낮추면 지연은 줄지만 요청 수는 늘어납니다.

다음은 test-topic과 client-id를 예시값으로 사용한 구성입니다.

type: streaming-kafka-export-task
config:
  maxParallelism: 1
  topic: test-topic
  clientId: client-id
  keyColumn: key
  valueColumn: value
  batchOptions:
    maxMilliseconds: 5000
    maxRecords: 1000
transforms:
  transformType: test
  userCodeMavenCoords: []
useDirectReaders: false

구성을 수정한 뒤 오른쪽 위의 Save로 저장합니다.

외부 변환에서 Kafka 사용

Python 외부 변환은 transforms-external-systems와 kafka-python을 사용하여 Kafka에 연결할 수 있습니다. 소스를 코드 저장소에 가져오고 해당 브로커에 대한 외부 통신 경로를 허용해야 합니다. 아래 코드는 SASL PLAIN과 TLS를 사용하는 구성이며 소스의 username, password 비밀 정보를 읽습니다. 다른 인증 방식의 클라이언트 설정을 그대로 대신하지는 않습니다.

각 예제는 별도 Python 변환 모듈입니다. 모듈 위쪽의 상수에 넣은 RID, 데이터셋 경로, 브로커와 토픽은 가상 예시값이므로 코드를 저장하기 전에 실제 소스와 리소스에 맞게 모두 수정합니다. 소스 RID는 Data Connection에서 가져온 소스의 RID를 사용합니다. 비밀번호는 상수에 넣지 않고 소스에서 읽습니다.

1분 또는 100건까지 읽기

이 예제는 JSON 객체 메시지를 읽어 pandas 테이블에 추가합니다. poll에도 남은 건수 제한을 전달하므로 한 번에 받은 배치 때문에 100건을 넘지 않습니다. 메시지가 없으면 빌드를 중단하여 빈 스키마를 쓰지 않습니다.

소비자 그룹의 자동 커밋은 Foundry 출력 확정과 별개입니다. Kafka 오프셋이 앞서 커밋된 뒤 변환이 실패하면 다음 실행에서 그 메시지를 건너뛸 수 있습니다. 새 그룹의 시작 위치는 클라이언트 기본값인 latest입니다. 재실행 위치를 Foundry에서 관리해야 한다면 다음 예제를 사용합니다.

import json
import time

import pandas as pd
from kafka import KafkaConsumer
from transforms.api import Output, incremental, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
OUTPUT_PATH = "/Examples/Kafka/json-events"
BOOTSTRAP_SERVERS = ["broker.example.com:9093"]
TOPIC = "example-json-events"
CONSUMER_GROUP = "foundry-json-example"


@external_systems(kafka_source=Source(SOURCE_RID))
@incremental()
@transform.using(output=Output(OUTPUT_PATH))
def compute(ctx, kafka_source, output):
    consumer = KafkaConsumer(
        TOPIC,
        bootstrap_servers=BOOTSTRAP_SERVERS,
        security_protocol="SASL_SSL",
        sasl_mechanism="PLAIN",
        sasl_plain_username=kafka_source.get_secret("username"),
        sasl_plain_password=kafka_source.get_secret("password"),
        group_id=CONSUMER_GROUP,
        enable_auto_commit=True,
        allow_auto_create_topics=False,
    )
    events = []
    deadline = time.monotonic() + 60
    try:
        while time.monotonic() < deadline and len(events) < 100:
            batches = consumer.poll(timeout_ms=1000, max_records=100 - len(events))
            for messages in batches.values():
                for message in messages:
                    event = json.loads(message.value)
                    if not isinstance(event, dict):
                        raise ValueError("이 예제의 메시지 본문은 JSON 객체여야 합니다.")
                    events.append(event)
    finally:
        consumer.close()
    if not events:
        ctx.abort_job()
        return
    output.write_table(pd.DataFrame(events))

이 예제의 출력 열은 JSON 필드에서 결정됩니다. 토픽의 메시지 스키마를 일정하게 유지해야 합니다. 미리보기에서도 실제 소비자 그룹의 오프셋이 변경될 수 있으므로 운영 그룹을 임의로 사용하지 않습니다.

Foundry 데이터셋에 오프셋 저장

이 방식은 Kafka 자동 커밋을 끄고 토픽의 파티션을 직접 할당합니다. 이전 상태의 offset + 1로 이동한 뒤 실행 시작 시점에 읽은 각 파티션의 끝 오프셋 직전까지만 처리합니다. 끝 오프셋은 마지막 메시지의 오프셋보다 1 큰 값입니다. 실행 중 새로 도착한 메시지는 다음 실행 범위에 남습니다.

records_output에는 메시지 행을 증분 추가하고 offsets_output에는 파티션별 마지막 처리 오프셋을 전체 교체 방식으로 저장합니다. 이번 실행에서 메시지를 받지 않은 파티션의 상태도 유지합니다. 같은 토픽을 여러 독립 변환이 처리하면 각자 전체 파티션을 읽으므로 이 코드를 소비자 그룹 기반의 작업 분산으로 해석하면 안 됩니다.

메시지의 키와 값은 UTF-8 문자열로 해석합니다. null과 빈 문자열은 구분합니다. 바이너리 페이로드를 다룰 때에는 디코딩 부분과 출력 자료형을 그 형식에 맞게 바꿔야 합니다. 토픽 보존 정책으로 저장한 재개 위치가 사라졌으면 조용히 최신 위치로 넘어가지 않고 실패합니다.

import polars as pl
from kafka import KafkaConsumer, TopicPartition
from transforms.api import Output, incremental, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
RECORDS_PATH = "/Examples/Kafka/records"
OFFSETS_PATH = "/Examples/Kafka/offsets"
BOOTSTRAP_SERVERS = ["broker.example.com:9093"]
TOPIC = "example-json-events"


RECORD_SCHEMA = {
    "topic": pl.String,
    "partition": pl.Int32,
    "offset": pl.Int64,
    "timestamp": pl.Int64,
    "key": pl.String,
    "value": pl.String,
}
OFFSET_SCHEMA = {"topic": pl.String, "partition": pl.Int32, "offset": pl.Int64}


@external_systems(kafka_source=Source(SOURCE_RID))
@incremental()
@transform.using(
    records_output=Output(RECORDS_PATH),
    offsets_output=Output(OFFSETS_PATH),
)
def compute(ctx, kafka_source, records_output, offsets_output):
    topic = TOPIC
    saved = {}
    if ctx.is_incremental:
        for row in offsets_output.polars(mode="previous").iter_rows(named=True):
            saved[(row["topic"], row["partition"])] = row["offset"]
    updated = dict(saved)
    rows = []
    consumer = KafkaConsumer(
        bootstrap_servers=BOOTSTRAP_SERVERS,
        security_protocol="SASL_SSL",
        sasl_mechanism="PLAIN",
        sasl_plain_username=kafka_source.get_secret("username"),
        sasl_plain_password=kafka_source.get_secret("password"),
        group_id=None,
        enable_auto_commit=False,
        auto_offset_reset="none",
        allow_auto_create_topics=False,
    )
    try:
        partition_ids = consumer.partitions_for_topic(topic)
        if not partition_ids:
            raise ValueError("토픽 파티션을 조회할 수 없습니다.")
        partitions = [TopicPartition(topic, p) for p in sorted(partition_ids)]
        consumer.assign(partitions)
        beginnings = consumer.beginning_offsets(partitions)
        ends = consumer.end_offsets(partitions)
        pending = set()
        for tp in partitions:
            last_offset = saved.get((tp.topic, tp.partition))
            start = beginnings[tp] if last_offset is None else last_offset + 1
            if not beginnings[tp] <= start <= ends[tp]:
                raise ValueError(f"저장한 오프셋이 현재 토픽 범위 밖에 있습니다: {tp}")
            consumer.seek(tp, start)
            if start < ends[tp]:
                pending.add(tp)
        inactive = set(partitions) - pending
        if inactive:
            consumer.pause(*inactive)
        idle_polls = 0
        while pending:
            batches = consumer.poll(timeout_ms=1000, max_records=1000)
            for tp, messages in batches.items():
                for message in messages:
                    if message.offset >= ends[tp]:
                        continue
                    rows.append({
                        "topic": message.topic,
                        "partition": message.partition,
                        "offset": message.offset,
                        "timestamp": message.timestamp,
                        "key": message.key.decode("utf-8") if message.key is not None else None,
                        "value": message.value.decode("utf-8") if message.value is not None else None,
                    })
                    updated[(tp.topic, tp.partition)] = message.offset
            finished = {tp for tp in pending if consumer.position(tp) >= ends[tp]}
            if finished:
                consumer.pause(*finished)
                pending.difference_update(finished)
            idle_polls = 0 if batches else idle_polls + 1
            if pending and idle_polls > 5:
                raise TimeoutError("끝 오프셋에 도달하기 전에 빈 폴링이 6회 이어졌습니다.")
    finally:
        consumer.close()
    if not rows:
        ctx.abort_job()
        return
    records_output.write_table(pl.DataFrame(rows, schema=RECORD_SCHEMA))
    checkpoint_rows = [
        {"topic": t, "partition": p, "offset": offset}
        for (t, p), offset in sorted(updated.items())
    ]
    offsets_output.set_mode("replace")
    offsets_output.write_table(pl.DataFrame(checkpoint_rows, schema=OFFSET_SCHEMA))

증분 실행에서는 메시지 출력의 기본 modify 모드가 기존 행에 새 행을 추가합니다. 비증분 실행에서는 기본 replace 모드를 사용하며 Kafka에 현재 보존된 시작 위치부터 다시 읽습니다. 두 출력은 이 변환에서 함께 관리해야 합니다. 보존 기간이 지난 Kafka 메시지까지 복원하는 예제는 아닙니다.

파티션 할당을 기다린다는 이유로 첫 poll의 메시지를 버리지 않습니다. 직접 할당과 seek를 먼저 수행합니다. 끝 오프셋에 도달하지 못한 상태에서 빈 폴링이 여섯 번 이어지면 성공으로 처리하지 않고 실패합니다. 메시지를 모두 메모리에 모으므로 대용량 토픽에서는 스키마를 정한 Parquet 파일에 버퍼를 나눠 기록하는 방식으로 변경하여 메모리 사용량을 제한해야 합니다.

가상 센서 메시지 전송

다음 예제는 실제 센서가 아닌 테스트용 데이터를 생성합니다. 온도는 평균 22°C와 표준편차 2, 습도는 평균 50과 표준편차 10, 기압은 평균 1013hPa와 표준편차 5인 정규분포를 사용합니다. 상태의 가중치는 OK 90, WARN 8, FAIL 2입니다.

대상 토픽에는 실제 쓰기가 발생합니다. 테스트용 토픽과 쓰기 권한을 준비해야 합니다. 약 1분 동안 전송하고 각 전송 확인 후 2초를 기다립니다. 전송 결과 Future를 확인한 행만 출력 기록에 포함합니다.

import json
import random
import time
from datetime import datetime, timezone

import pandas as pd
from kafka import KafkaProducer
from transforms.api import Output, transform
from transforms.external.systems import Source, external_systems

SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
OUTPUT_PATH = "/Examples/Kafka/sensor-send-results"
BOOTSTRAP_SERVERS = ["broker.example.com:9093"]
TOPIC = "example-sensor-test"


def sensor_sample():
    return {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "temperature_c": round(random.gauss(22, 2), 2),
        "humidity_percent": round(random.gauss(50, 10), 1),
        "pressure_hpa": round(random.gauss(1013, 5), 1),
        "status": random.choices(["OK", "WARN", "FAIL"], weights=[90, 8, 2], k=1)[0],
    }


@external_systems(kafka_source=Source(SOURCE_RID))
@transform.using(output=Output(OUTPUT_PATH))
def compute(kafka_source, output):
    producer = KafkaProducer(
        bootstrap_servers=BOOTSTRAP_SERVERS,
        security_protocol="SASL_SSL",
        sasl_mechanism="PLAIN",
        sasl_plain_username=kafka_source.get_secret("username"),
        sasl_plain_password=kafka_source.get_secret("password"),
        acks="all",
    )
    sent = []
    deadline = time.monotonic() + 60
    try:
        while time.monotonic() < deadline:
            sample = sensor_sample()
            metadata = producer.send(
                TOPIC,
                value=json.dumps(sample).encode("utf-8"),
            ).get(timeout=30)
            sent.append({**sample, "partition": metadata.partition, "offset": metadata.offset})
            time.sleep(2)
        producer.flush()
    finally:
        producer.close()
    output.write_table(pd.DataFrame(sent))

전송 확인 대기 시간이 초과되어도 브로커가 메시지를 나중에 받을 수 있습니다. Foundry 빌드 실패나 재실행이 외부 Kafka 쓰기를 롤백하지도 않습니다. 프로듀서의 멱등성만으로 변환 전체 재실행의 중복이 제거된다고 가정하면 안 됩니다.