자주 묻는 질문(FAQ)

Azure Databricks 함께 Kafka를 사용하는 방법에 대한 일반적인 질문입니다.

Kafka 옵션이 지원되지 않거나 인식되지 않는다는 오류가 발생하는 이유는 무엇인가요?

이 오류는 Kafka 클라이언트 구성 옵션을 설정할 때 접두사를 kafka. 사용하지 않은 경우에 발생합니다. Kafka 클라이언트에 직접 전달되는 모든 옵션 앞에는 다음 접두 kafka.사를 지정해야 합니다.

다음 코드에서는 접두사를 누락 kafka. 한 잘못된 옵션을 보여 있습니다.

.option("security.protocol", "SASL_SSL")
.option("sasl.mechanism", "PLAIN")

다음 코드는 올바른 옵션을 보여줍니다.

.option("kafka.security.protocol", "SASL_SSL")
.option("kafka.sasl.mechanism", "PLAIN")

Spark Kafka 커넥터(예: subscribe, startingOffsetsmaxOffsetsPerTrigger)에 대한 옵션은 접두사를 필요로 하지 않습니다. 전체 옵션 목록은 Kafka를 참조하세요.

음영 처리된 Kafka 클래스에 대한 오류가 표시되는 이유는 무엇인가요?

Azure Databricks는 특수 처리된 Kafka 클래스(접두사로 kafkashaded. 또는 shadedmskiam.)를 사용해야 합니다. 다음과 같은 RESTRICTED_STREAMING_OPTION_PERMISSION_ENFORCED오류가 표시되면 음영 처리된 클래스 이름을 사용해야 합니다.

org.apache.kafka.* 클래스에는 접두사가 kafkashaded. 필요합니다. 예: kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule
software.amazon.msk.* 클래스에는 접두사가 shadedmskiam. 필요합니다. 예: shadedmskiam.software.amazon.msk.auth.iam.IAMLoginModule

Kafka에 연결할 때 `TimeoutException`이 나오는 이유는 무엇인가요?

일반적인 원인은 다음과 같습니다.

네트워크 연결: 컴퓨팅 클러스터가 Kafka broker에 연결할 수 없습니다. 방화벽 규칙, 보안 그룹 및 VPC 구성을 확인합니다.
잘못된 부트스트랩 서버: 호스트 이름과 포트가 kafka.bootstrap.servers 올바른지 확인합니다.
DNS 확인: Azure Databricks 네트워크에서 Kafka broker 호스트 이름을 확인할 수 있는지 확인합니다.
SSL/TLS 문제: SSL을 사용하는 경우 인증서가 올바르게 구성되었는지 확인합니다.

Private Link 또는 VPC 피어링 설정의 경우 올바른 네트워크 경로가 있는지 확인합니다.

Kafka에 일괄 처리 또는 스트리밍 모드를 사용해야 하나요?

사용 사례에 따라 달라집니다.

스트리밍 모드 (spark.readStream): 연속 데이터 처리 또는 짧은 대기 시간 수집이 필요한 경우에 사용합니다.
일괄 처리 모드 (spark.read): 일회성 데이터 로드, 백필 또는 디버깅에 사용합니다. 둘 다 startingOffsets 및 endingOffsets.가 필요합니다.

및 실시간 모드와 같은 AvailableNowProcessingTime트리거 간격을 구성하는 방법에 대한 자세한 내용은 구조적 스트리밍 트리거 간격 구성을 참조하세요.

단일 스트림에서 여러 Kafka 토픽을 읽을 수 있나요?

예, 다음을 사용할 수 있습니다.

subscribe: 쉼표로 구분된 항목 목록을 제공합니다. 예를 들면 다음과 같습니다 .option("subscribe", "topic1,topic2").
subscribePattern: Java 정규식 패턴을 사용하여 토픽 이름과 일치합니다(예: .option("subscribePattern", "topic-.*")).

Lakeflow Spark 선언적 파이프라인에서 Kafka를 사용하려면 어떻게 해야 하나요?

Lakeflow Spark 선언적 파이프라인은 Kafka 원본을 기본적으로 지원합니다.

다음 코드와 같이 Kafka에서 읽는 스트리밍 테이블을 정의할 수 있습니다.

Python

import dlt

@dlt.table
def kafka_bronze():
  return (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<server:port>")
    .option("subscribe", "<topic>")
    .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE kafka_bronze AS
SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<server:port>',
  subscribe => '<topic>'
);

Lakeflow Spark 선언적 파이프라인의 스트리밍 원본에 대한 자세한 내용은 파이프라인의 데이터 로드 를 참조하세요.

Kafka 키 및 값 열을 역직렬화하려면 어떻게 해야 하나요?

key 열과 value 열은 BINARY 형식으로 반환됩니다. 데이터 프레임 작업을 사용하여 데이터 형식에 따라 역직렬화합니다.

문자열 데이터: 이진 파일을 문자열로 변환하는 데 사용합니다 cast("string") .
JSON 데이터: 문자열로 캐스팅한 후 from_json() 를 사용합니다. from_json 함수참조하세요.
Avro 데이터: Avro로 인코딩된 데이터를 역직렬화하는 데 사용합니다 from_avro() . 스트리밍 Avro 데이터 읽기 및 쓰기를 참조하세요.
프로토콜 버퍼: protobuf 데이터를 역직렬화하는 데 사용합니다 from_protobuf() . 읽기 및 쓰기 프로토콜 버퍼를 참조하세요.

idempotent 쓰기 오류가 표시되는 이유는 무엇인가요?

Databricks Runtime 13.3 LTS 이상에는 기본적으로 idempotent 쓰기를 사용하도록 설정하는 최신 버전의 kafka-clients 라이브러리가 포함되어 있습니다. Kafka 클러스터가 버전 2.8.0 이하를 사용하고 있으며, ACL이 구성되었지만 IDEMPOTENT_WRITE이 활성화되지 않은 경우, org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state로 인해 쓰기가 실패합니다.

이 오류를 해결하려면 Kafka 버전 2.8.0 이상으로 업그레이드하거나 구조적 스트림 기록기를 구성할 때 .option("kafka.enable.idempotence", "false")를 설정하세요.

`KAFKA_DATA_LOSS_ERROR` 무엇이며 어떻게 해결합니까?

이 오류는 Kafka 원본이 검사점에 저장된 오프셋을 Kafka에서 더 이상 사용할 수 없다는 것을 감지할 때 발생합니다. 일반적으로 다음과 같은 이유로 발생합니다.

스트림이 Kafka 보존 기간보다 더 오래 일시 중지되었습니다.
Kafka 토픽 데이터가 삭제되었거나 토픽이 다시 생성되었습니다.
Kafka broker는 데이터 손실을 경험했습니다.

해결하려면:

데이터 손실이 허용되는 경우: 스트림이 사용 가능한 가장 빠른 오프셋에서 계속되도록 설정합니다 .option("failOnDataLoss", "false") .
데이터 손실이 허용되지 않는 경우: 검사점을 다시 설정하여 오프셋에서 earliest 다시 처리하거나 누락된 Kafka 데이터를 복원합니다.

자세한 내용은 KAFKA_DATA_LOSS 오류 조건을 참조하세요.

Kafka에서 데이터를 읽는 속도를 제어하려면 어떻게 해야 하나요?

maxOffsetsPerTrigger 마이크로 일괄 처리당 처리되는 오프셋 수(약 레코드 수)를 제한하려면 이 옵션을 사용합니다. 이렇게 하면 백로그를 따라잡을 때 다운스트림 처리를 압도하거나 메모리 문제를 일으킬 수 있는 대규모 일괄 처리를 방지할 수 있습니다.

Python

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:port>")
  .option("subscribe", "<topic>")
  .option("maxOffsetsPerTrigger", 10000)
  .load()
)

Scala

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:port>")
  .option("subscribe", "<topic>")
  .option("maxOffsetsPerTrigger", 10000)
  .load()

SQL

SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<server:port>',
  subscribe => '<topic>',
  maxOffsetsPerTrigger => '10000'
);

또는 각 일괄 처리에 대해 생성되는 Spark 파티션 수를 제어하거나 minPartitions 같은 maxRecordsPerPartition 옵션을 사용합니다.

최신 Kafka 오프셋에서 스트림이 얼마나 뒤처지는지 모니터링하려면 어떻게 해야 하나요?

스트리밍 쿼리 진행률에서 사용할 수 있는 avgOffsetsBehindLatest, maxOffsetsBehindLatest, minOffsetsBehindLatest 메트릭을 사용합니다. 이러한 보고서는 구독한 모든 토픽 파티션에서 스트림이 최신 사용 가능한 오프셋에서 얼마나 뒤처져 있는지를 보고합니다. Azure Databricks의 구조적 스트리밍 쿼리 모니터링을 참조하세요.

아직 처리되지 않은 데이터의 총 바이트를 추정하는 데 사용할 estimatedTotalBytesBehindLatest 수도 있습니다.

Databricks Runtime 17.1로 업그레이드한 후 Kafka 오프셋 지연 메트릭이 0이 아닌 영구 값을 표시하는 이유는 무엇인가요?

Databricks Runtime 17.1 이상에서는 각 마이크로 일괄 처리가 완료된 후 최신 Kafka 오프셋을 가져옵니다. 데이터를 지속적으로 받는 토픽에서 백로그 메트릭은 0이 아닌 작은 값을 표시할 수 있습니다. 이는 예상되는 동작이며 스트림이 뒤처지고 있음을 나타내지 않습니다.

Databricks Runtime 17.0 이하에서는 마이크로 일괄 처리 시작 시간에 최신 Kafka 오프셋을 가져옵니다. 스트리밍 쿼리가 마이크로 일괄 처리 시작 시 사용 가능한 모든 레코드를 일관되게 사용하는 경우 백로그 메트릭이 반환 0 할 수 있습니다.

값이 크거나 지속적으로 증가하는 경우 스트림이 들어오는 데이터를 유지하지 못할 수 있습니다. Azure Databricks의 구조적 스트리밍 쿼리 모니터링을 참조하세요.

내 Kafka 스트림 초기화 속도가 느린 이유는 무엇인가요?

Kafka 스트림은 다음을 수행할 시간이 필요합니다.

Kafka 클러스터에 연결하고 메타데이터를 가져옵니다.
토픽 파티션을 검색합니다.
초기 오프셋을 가져옵니다.

온-프레미스 또는 원격 Kafka 클러스터의 경우 네트워크 대기 시간이 초기화 시간에 큰 영향을 미칠 수 있습니다. 자주 다시 시작하는 트리거/예약된 파이프라인을 실행하는 경우 반복적인 초기화 오버헤드를 방지하기 위해 연속 스트리밍 모드를 사용하는 것이 좋습니다.

Spark 실행기를 더 추가해도 Kafka 처리량이 증가하지 않는 이유는 무엇인가요?

Kafka broker가 포화 상태가 되면 Spark 실행기를 추가하면 처리량을 늘리지 않고 비용이 증가합니다.

Kafka가 병목 상태임을 표시합니다.

더 많은 코어를 추가했음에도 불구하고 처리량은 정체되었습니다.
Kafka broker CPU 또는 네트워크 사용률이 높습니다.
Spark 작업은 신속하게 완료되지만 새 데이터를 기다립니다.

이 문제를 해결하려면 Broker를 추가하거나 파티션 수를 늘려 부하를 분산하여 Kafka 클러스터의 크기를 조정합니다.

Kafka 스트리밍에 대한 비용 및 컴퓨팅 사용률을 최적화하는 방법은 무엇입니까?

마이크로 일괄 처리 및 AvailableNow 모드의 경우:

클러스터 크기 조정: 메트릭을 모니터링하고 최대 부하에 적절한 고정 클러스터 크기를 설정합니다.
사용 maxOffsetsPerTrigger: 부하 급증 시 리소스 사용량을 제어하도록 일괄 처리 크기를 제한합니다.
자동 크기 조정 방지: 스트리밍 작업이 지속적으로 실행되고 노드를 추가하거나 제거하면 태스크가 오버헤드를 재조정합니다.
데이터 기울이기 감소: 왜곡된 파티션으로 인해 일부 작업이 다른 작업보다 훨씬 더 많은 데이터를 처리하게 되며, 이로 인해 전체 일괄 처리 완료 속도가 느려지고 유휴 작업에서 컴퓨팅 리소스가 낭비됩니다. minPartitions 보다 균형 잡힌 처리를 위해 큰 Kafka 파티션을 더 작은 Spark 파티션으로 분할하는 옵션을 사용합니다.

실시간 모드의 경우 데이터를 기다리는 동안 작업이 유휴 상태를 유지할 수 있으므로 컴퓨팅 크기 조정이 특히 중요합니다. 주요 고려 사항:

각 태스크가 여러 Kafka 파티션을 처리하여 오버헤드를 줄이도록 설정합니다 maxPartitions .
Shuffle 작업이 많은 작업을 최적화 spark.sql.shuffle.partitions 합니다.

실시간 모드의 클러스터 크기 조정에 대한 지침은 컴퓨팅 크기 조정 을 참조하세요.

토픽에 데이터가 있더라도 내 스트림이 레코드를 반환하지 않는 이유는 무엇인가요?

일반적인 원인은 다음과 같습니다.

잘못된 startingOffsets 설정: 기본값은 스트림이 latest시작된 후에 도착하는 새 데이터만 읽는 것입니다. startingOffsets를 earliest로 설정하여 기존 데이터를 읽습니다.
잘못된 토픽 이름: 올바른 토픽을 구독하고 있는지 확인합니다.
인증 문제: 스트림이 성공적으로 연결되었을 수 있지만 토픽에서 읽을 수 있는 권한이 없습니다. Kafka ACL을 확인합니다.
오프셋 만료: 스트림이 오랫동안 중지되고 검사점의 오프셋이 만료된 경우(Kafka 보존에 의해 삭제됨) 검사점을 다시 설정하거나 조정 failOnDataLoss해야 할 수 있습니다.

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-05-27

자주 묻는 질문(FAQ)

Kafka 옵션이 지원되지 않거나 인식되지 않는다는 오류가 발생하는 이유는 무엇인가요?

음영 처리된 Kafka 클래스에 대한 오류가 표시되는 이유는 무엇인가요?

Kafka에 연결할 때 TimeoutException이 나오는 이유는 무엇인가요?

Kafka에 일괄 처리 또는 스트리밍 모드를 사용해야 하나요?

단일 스트림에서 여러 Kafka 토픽을 읽을 수 있나요?

Lakeflow Spark 선언적 파이프라인에서 Kafka를 사용하려면 어떻게 해야 하나요?

Python

SQL

Kafka 키 및 값 열을 역직렬화하려면 어떻게 해야 하나요?

idempotent 쓰기 오류가 표시되는 이유는 무엇인가요?

KAFKA_DATA_LOSS_ERROR 무엇이며 어떻게 해결합니까?

Kafka에서 데이터를 읽는 속도를 제어하려면 어떻게 해야 하나요?

Python

Scala

SQL

최신 Kafka 오프셋에서 스트림이 얼마나 뒤처지는지 모니터링하려면 어떻게 해야 하나요?

Databricks Runtime 17.1로 업그레이드한 후 Kafka 오프셋 지연 메트릭이 0이 아닌 영구 값을 표시하는 이유는 무엇인가요?

내 Kafka 스트림 초기화 속도가 느린 이유는 무엇인가요?

Spark 실행기를 더 추가해도 Kafka 처리량이 증가하지 않는 이유는 무엇인가요?

Kafka 스트리밍에 대한 비용 및 컴퓨팅 사용률을 최적화하는 방법은 무엇입니까?

토픽에 데이터가 있더라도 내 스트림이 레코드를 반환하지 않는 이유는 무엇인가요?

피드백

추가 리소스

Kafka에 연결할 때 `TimeoutException`이 나오는 이유는 무엇인가요?

`KAFKA_DATA_LOSS_ERROR` 무엇이며 어떻게 해결합니까?