본문 바로가기
WIKI 기술 지식 베이스

Kafka source

원문 보기 위키 갱신

출처: 문서

본문

Kafka 소스

OpenSearch Data Prepper에서는 Apache Kafka 소스(kafka)를 사용해 하나 이상의 Kafka 토픽에서 레코드를 읽을 수 있어요. 이 레코드에는 Data Prepper 파이프라인이 수집할 수 있는 이벤트가 담겨 있어요. kafka 소스는 Kafka의 Consumer API를 사용해 Kafka 브로커에서 메시지를 소비하고, 이어서 Data Prepper 이벤트를 만들어 Data Prepper 파이프라인이 후속 처리를 하도록 해요.

사용법

다음 예시는 Data Prepper 파이프라인에서 kafka 소스를 사용하는 모습이에요:

kafka-pipeline:
  source:
    kafka:
      bootstrap_servers:
        - 127.0.0.1:9092
      topics:
        - name: Topic1
          group_id: groupID1
        - name: Topic2
          group_id: groupID1
  sink:
    - stdout: {}

설정

kafka 소스에는 다음 설정 옵션을 사용해요.

옵션 필수 여부 타입 설명
bootstrap_servers 예. 클러스터로 Amazon Managed Streaming for Apache Kafka(Amazon MSK)를 사용하지 않을 때. IP address Kafka 클러스터에 초기 연결을 위한 호스트 또는 포트예요. 각 브로커의 IP 주소나 포트 번호를 사용해 여러 Kafka 브로커를 설정할 수 있어요. Amazon MSK를 Kafka 클러스터로 사용할 때는 설정에 제공된 MSK Amazon Resource Name(ARN)을 통해 MSK에서 부트스트랩 서버 정보를 가져와요.
topics 예 JSON array Data Prepper kafka 소스가 메시지를 읽는 데 사용하는 Kafka 토픽이에요. 토픽은 최대 10개까지 설정할 수 있어요. topics 설정 옵션에 대한 자세한 내용은 토픽을 참고하세요.
schema 아니요 JSON object 스키마 레지스트리 설정이에요. 자세한 내용은 스키마를 참고하세요.
authentication 아니요 JSON object 파이프라인과 Kafka 양쪽의 인증 옵션을 설정해요. 자세한 내용은 인증을 참고하세요.
encryption 아니요 JSON object 암호화 설정이에요. 자세한 내용은 암호화를 참고하세요.
aws 아니요 JSON object AWS 설정이에요. 자세한 내용은 aws를 참고하세요.
acknowledgments 아니요 Boolean true이면 OpenSearch 싱크가 이벤트를 받았을 때 kafka 소스가 엔드 투 엔드 승인을 받을 수 있어요. 기본값은 false예요.
client_dns_lookup 예. DNS 별칭을 사용할 때. String Kafka의 client.dns.lookup 옵션을 설정해요. 기본값은 default예요.

토픽

topics 배열에서는 토픽마다 다음 옵션을 사용해요.

옵션 필수 여부 타입 설명
name 예 String 각 Kafka 토픽의 이름이에요.
group_id 예 String Kafka의 group.id 옵션을 설정해요.
workers 아니요 Integer 각 토픽에 연결되는 멀티스레드 컨슈머 수예요. 기본값은 2이고 최댓값은 200이에요.
serde_format 아니요 String 토픽에 있는 메시지의 직렬화 및 역직렬화 형식을 나타내요. 기본값은 plaintext예요.
auto_commit 아니요 Boolean false이면 컨슈머 오프셋이 백그라운드에서 주기적으로 Kafka에 커밋되지 않아요. 기본값은 false예요.
commit_interval 아니요 Integer auto_commit이 true로 설정된 경우, Kafka의 auto.commit.interval.ms 옵션을 통해 컨슈머 오프셋을 Kafka에 자동 커밋하는 주기를 초 단위로 설정해요. 기본값은 5s예요.
session_timeout 아니요 Integer Kafka의 그룹 관리 기능을 사용할 때 소스가 클라이언트 장애를 감지하는 시간이에요. 이 기능은 데이터 스트림의 밸런스를 맞추는 데 사용할 수 있어요. 기본값은 45s예요.
auto_offset_reset 아니요 String Kafka의 auto.offset.reset 옵션을 통해 오프셋을 이전 오프셋 또는 최신 오프셋으로 자동 재설정해요. 기본값은 earliest예요.
thread_waiting_time 아니요 Integer 스레드가 앞선 스레드가 작업을 마치고 다음 스레드에 신호를 보낼 때까지 기다리는 시간이에요. Kafka 컨슈머 API의 poll 타임아웃 값은 이 설정의 절반으로 설정돼요. 기본값은 5s예요.
max_partition_fetch_bytes 아니요 Integer Kafka의 max.partition.fetch.bytes 설정을 통해 각 파티션에서 반환되는 최대 데이터량을 메가바이트 단위로 설정해요. 기본값은 1mb예요.
heart_beat_interval 아니요 Integer Kafka의 heartbeat.interval.ms 설정을 통해 그룹 관리 기능을 사용할 때 컨슈머 코디네이터로 하트비트를 보내는 예상 간격이에요. 기본값은 5s예요.
fetch_max_wait 아니요 Integer fetch_min_bytes 요구 사항을 충족할 만큼 데이터가 충분하지 않을 때 서버가 페치 요청을 블록하는 최대 시간이에요. Kafka의 fetch.max.wait.ms 설정을 통해 지정해요. 기본값은 500ms예요.
fetch_max_bytes 아니요 Integer Kafka의 fetch.max.bytes 설정을 통해 브로커가 수용하는 최대 레코드 크기예요. 기본값은 50mb예요.
fetch_min_bytes 아니요 Integer 페치 요청 시 서버가 반환하는 최소 데이터 양이에요. Kafka의 retry.backoff.ms 설정을 통해 지정해요. 기본값은 1b예요.
retry_backoff 아니요 Integer 특정 토픽 파티션에 대해 실패한 요청을 재시도하기 전에 기다리는 시간이에요. 기본값은 10s예요.
max_poll_interval 아니요 Integer Kafka의 max.poll.interval.ms 옵션을 통해 그룹 관리를 사용할 때 poll() 호출 사이의 최대 지연 시간이에요. 기본값은 300s예요.
consumer_max_poll_records 아니요 Integer Kafka의 max.poll.records 설정을 통해 한 번의 poll() 호출에서 반환되는 최대 레코드 수예요. 기본값은 500이에요.
key_mode 아니요 String Kafka 메시지의 키 필드를 처리하는 방식을 나타내요. 기본 설정은 include_as_field로, 키를 kafka_key 이벤트에 포함해요. include_as_metadata로 설정하면 키를 이벤트의 메타데이터에 포함하고, discard로 설정하면 키를 버려요.

스키마

schema 설정에는 다음 옵션이 있어요.

옵션 타입 필수 여부 설명
type String 예 레지스트리에 맞춰 스키마 타입을 설정해요. 유효한 값은 aws_glue(AWS Glue 스키마 레지스트리)와 confluent(Confluent 스키마 레지스트리)예요. aws_glue 레지스트리를 사용할 때는 AWS 설정 옵션을 함께 설정하세요.
basic_auth_credentials_source String 아니요 스키마 레지스트리 자격 증명을 가져오는 위치예요. api_key/api_secret을 제공할 때는 USER_INFO를 사용해요. 그 외 유효한 값은 URL과 SASL_INHERIT이에요. 기본값은 일반적으로 기반 클라이언트의 설정을 따라요.

다음 설정 옵션은 confluent 레지스트리를 사용할 때만 필요해요.

옵션 타입 설명
registry_url String 스키마 레지스트리의 기본 URL이에요(예: http://schema-registry:8081 또는 https://sr.example.com).
version String 서브젝트별로 사용할 스키마 버전이에요. 정수 또는 latest를 사용해요.
api_key String 스키마 레지스트리 API 키예요.
api_secret String 스키마 레지스트리 API 시크릿이에요.

다음 예시는 스키마 레지스트리를 설정하는 방법이에요:

schema:
  type: confluent
  registry_url: "http://schema-registry:8081"
  api_key: "<optional if using basic/key auth>"
  api_secret: "<optional if using basic/key auth>"
  version: "latest"

{% include copy.html %}

TLS를 통한 스키마 레지스트리

Kafka 소스는 https로 스키마 레지스트리에 연결할 때 JVM 트러스트스토어를 사용해요. 스키마 레지스트리가 사용자 지정 CA로 서명된 경우에는 해당 CA를 Data Prepper JVM 트러스트스토어에 추가하거나, 환경 변수를 사용해 사용자 지정 트러스트스토어를 제공하세요.

다음 명령으로 CA 인증서를 담은 트러스트스토어를 만들 수 있어요:

keytool -importcert -noprompt -alias sr-ca -file sr-ca.pem -keystore /usr/share/data-prepper/certs/sr.truststore.jks -storepass changeit

{% include copy.html %}

다음 명령은 JAVA_TOOL_OPTIONS를 사용해 Data Prepper를 설정해요:

JAVA_TOOL_OPTIONS=-Djavax.net.ssl.trustStore=/usr/share/data-prepper/certs/sr.truststore.jks -Djavax.net.ssl.trustStorePassword=changeit

{% include copy.html %}

다음 방법으로 docker-compose.yaml에서 Data Prepper를 설정할 수 있어요:

environment:
  - JAVA_TOOL_OPTIONS=-Djavax.net.ssl.trustStore=/usr/share/data-prepper/certs/sr.truststore.jks -Djavax.net.ssl.trustStorePassword=changeit
volumes:
  - ./certs:/usr/share/data-prepper/certs:ro

{% include copy.html %}

인증

authentication 섹션은 SASL을 설정해요:

authentication:
  sasl:
    plaintext:
      username: alice
      password: secret

{% include copy.html %}

옵션 타입 설명
sasl Object SASL 설정이에요.

SASL

SASL 인증을 설정할 때는 다음 옵션 중 하나를 사용해요.

옵션 타입 설명
plaintext JSON object 플레인텍스트 인증 설정이에요. 이전 버전과의 호환성을 위해 plain 별칭도 지원해요. 자세한 내용은 SASL plaintext를 참고하세요.
aws_msk_iam String Amazon MSK의 AWS Identity and Access Management(IAM) 설정이에요. role로 설정하면 aws 설정에서 지정한 sts_role_arm을 사용해요. 기본값은 default예요.
SASL plaintext

SASL.plain 프로토콜을 사용할 때는 다음 옵션이 필요해요.

옵션 타입 설명
username String SASL/PLAIN 사용자 이름이에요.
password String SASL/PLAIN 비밀번호예요.

암호화

SSL 암호화를 설정할 때는 다음 옵션을 사용해요.

옵션 필수 여부 타입 설명
type 아니요 String 암호화 타입이에요. 암호화를 비활성화하려면 none을 사용해요. 기본값은 ssl이에요.
certificate 아니요 String SSL 인증서 내용이에요. 이 옵션과 trust_store_file_path 중 하나만 사용하고 둘을 함께 사용하지는 마세요.
trust_store_file_path 아니요 String SSL 인증서가 들어 있는 트러스트스토어 파일의 경로예요. 이 옵션과 certificate 중 하나만 사용하고 둘을 함께 사용하지는 마세요.
trust_store_password 아니요 String 트러스트스토어 파일의 비밀번호예요.
insecure 아니요 Boolean SSL 인증서 검증을 끄는 데 사용하는 불리언 플래그예요. true로 설정하면 CA(인증 기관) 인증서 검증이 꺼지고 보안되지 않은 HTTP 요청이 전송돼요. 기본값은 false예요.

다음 설정을 사용하면 SSL 암호화를 활성화할 수 있어요:

encryption:
  type: ssl
  # With public CA: no extra config needed.
  # With private CA: trust using JVM truststore.

{% include copy.html %}

AWS

aws 서비스의 인증을 설정할 때는 다음 옵션을 사용해요.

옵션 필수 여부 타입 설명
region 아니요 String 자격 증명에 사용할 AWS 리전이에요. 기본값은 리전을 결정하는 표준 SDK 동작을 따라요.
sts_role_arn 아니요 String Amazon Simple Queue Service(Amazon SQS)와 Amazon Simple Storage Service(Amazon S3)에 요청을 보낼 때 가정하는 AWS Security Token Service(AWS STS) 역할이에요. 기본값은 null이며, 이 경우 자격 증명에 대한 표준 SDK 동작을 사용해요.
msk 아니요 JSON object MSK 설정이에요.

MSK

msk 객체 안에서는 다음 옵션을 사용해요.

옵션 필수 여부 타입 설명
arn 예 String 사용

더 알아보기 (Learn more)

함께 읽는 실무 이야기

우리 조직에도 적용하고 싶다면, AI 시스템 구축 서비스를 살펴보세요.