Kafka 연결 - AWS Glue
Kafka 구성Kafka에서 읽기Kafka에 쓰기Kafka 연결 파라미터

Kafka 연결

Data Catalog 테이블에 저장된 정보를 사용하거나 데이터 스트림에 직접 액세스할 수 있는 정보를 제공하여 Kafka 연결을 통해 Kafka Data Streams에서 읽고 쓸 수 있습니다. 연결에서는 Kafka 클러스터 또는 Amazon Managed Streaming for Apache Kafka 클러스터를 지원합니다. Kafka의 정보를 Spark DataFrame으로 읽은 다음 AWS Glue DynamicFrame으로 변환할 수 있습니다. DynamicFrame을 JSON 형식으로 Kafka에 쓸 수 있습니다. 데이터 스트림에 직접 액세스하는 경우 이러한 옵션을 사용하여 데이터 스트림에 액세스하는 방법에 대한 정보를 제공합니다.

getCatalogSource 또는 create_data_frame_from_catalog를 사용하여 Kafka 스트리밍 소스에서 레코드를 소비하거나 getCatalogSink 또는 write_dynamic_frame_from_catalog를 사용하여 Kafka에 레코드를 쓰는 경우 작업에는 데이터 카탈로그 데이터베이스와 테이블 이름 정보가 있으며 이를 사용하여 Kafka 스트리밍 소스에서 읽기 위한 몇 가지 기본 파라미터를 얻을 수 있습니다. getSource, getCatalogSink, getSourceWithFormat, getSinkWithFormat, createDataFrameFromOptions, create_data_frame_from_options 또는 write_dynamic_frame_from_catalog를 사용하는 경우 여기에 설명된 연결 옵션을 통해 이러한 기본 파라미터를 지정해야 합니다.

GlueContext 클래스에 지정된 메서드에 대해 다음 인수를 사용하여 Kafka에 대한 연결 옵션을 지정할 수 있습니다.

  • Scala

    • connectionOptions: getSource, createDataFrameFromOptions, getSink와(과) 함께 사용

    • additionalOptions: getCatalogSource, getCatalogSink와 함께 사용

    • options: getSourceWithFormat, getSinkWithFormat와 함께 사용

  • Python

    • connection_options: create_data_frame_from_options, write_dynamic_frame_from_options와 함께 사용

    • additional_options: create_data_frame_from_catalog, write_dynamic_frame_from_catalog와 함께 사용

    • options: getSource, getSink와 함께 사용

스트리밍 ETL 작업에 대한 참고 및 제한 사항은 스트리밍 ETL 참고 사항 및 제한 사항 섹션을 참조하세요.

주제

    Kafka 구성

    인터넷을 통해 사용할 수 있는 Kafka 스트림에 연결하기 위한 AWS의 필수 조건은 없습니다.

    AWS Glue Kafka 연결을 생성하여 연결 보안 인증을 관리할 수 있습니다. 자세한 내용은 Apache Kafka 데이터 스트림에 대한 AWS Glue 연결 생성 단원을 참조하십시오. AWS Glue 작업 구성에서 추가 네트워크 연결로 connectionName을 제공하고 메서드 직접 호출에서 connectionName 파라미터로 ConnectionName을 제공합니다.

    경우에 따라 추가 필수 조건을 구성해야 합니다.

    • IAM 인증과 함께 Amazon Managed Streaming for Apache Kafka를 사용하는 경우 적절한 IAM 구성이 필요합니다.

    • Amazon VPC 내에서 Amazon Managed Streaming for Apache Kafka를 사용하는 경우 적절한 Amazon VPC 구성이 필요합니다. Amazon VPC 연결 정보를 제공하는 AWS Glue 연결을 생성해야 합니다. AWS Glue 연결을 추가 네트워크 연결로 포함하려면 작업 구성이 필요합니다.

    스트리밍 ETL 작업 필수 조건에 대한 자세한 내용은 AWS Glue에서 스트리밍 ETL 작업 섹션을 참조하세요.

    예제: Kafka 스트림에서 읽기

    forEachBatch과(와) 함께 사용합니다.

    Kafka 스트리밍 소스의 예:

    kafka_options =
    { "connectionName": "ConfluentKafka", 
      "topicName": "kafka-auth-topic", 
      "startingOffsets": "earliest", 
      "inferSchema": "true", 
      "classification": "json" 
    }
data_frame_datasource0 = glueContext.create_data_frame.from_options(connection_type="kafka", connection_options=kafka_options)

    예: Kafka 스트림에 쓰기

    Kafka에 쓰기의 예제:

    getSink 메서드를 사용한 예제:

    data_frame_datasource0 = 
glueContext.getSink(
	connectionType="kafka",
	connectionOptions={
		JsonOptions("""{
			"connectionName": "ConfluentKafka", 
			"classification": "json", 
			"topic": "kafka-auth-topic", 
			"typeOfData": "kafka"}
		""")}, 
	transformationContext="dataframe_ApacheKafka_node1711729173428")
	.getDataFrame()

    write_dynamic_frame.from_options 메서드를 사용한 예제:

    kafka_options =
    { "connectionName": "ConfluentKafka", 
      "topicName": "kafka-auth-topic", 
      "classification": "json" 
    }
data_frame_datasource0 = glueContext.write_dynamic_frame.from_options(connection_type="kafka", connection_options=kafka_options)

    Kafka 연결 옵션 참조

    읽을 때는 "connectionType": "kafka"를 사용한 다음 연결 옵션을 사용합니다.

    • "bootstrap.servers" (필수) 부트스트랩 서버 URL의 목록입니다(예: b-1.vpc-test-2.o4q88o.c6.kafka.us-east-1.amazonaws.com:9094). 이 옵션은 API 호출에 지정하거나 데이터 카탈로그의 테이블 메타데이터에 정의해야 합니다.

    • "security.protocol" (필수) 브로커와 통신하는 데 사용되는 프로토콜입니다. 가능한 값은 "SSL" 또는 "PLAINTEXT"입니다.

    • "topicName": (필수) 쉼표로 구분된 구독할 주제의 목록입니다. "topicName", "assign" 또는 "subscribePattern" 중에서 하나만 지정해야 합니다.

    • "assign": (필수) 사용할 특정 TopicPartitions을 지정하는 JSON 문자열입니다. "topicName", "assign" 또는 "subscribePattern" 중에서 하나만 지정해야 합니다.

      예: '{"topicA":[0,1],"topicB":[2,4]}'

    • "subscribePattern": (필수) 구독할 주제 목록을 식별하는 Java 정규식 문자열입니다. "topicName", "assign" 또는 "subscribePattern" 중에서 하나만 지정해야 합니다.

      예: 'topic.*'

    • "classification"(필수) 레코드의 데이터에서 사용하는 파일 형식입니다. 데이터 카탈로그를 통해 제공되지 않는 한 필수입니다.

    • "delimiter"(선택 사항) classification이 CSV일 때 사용되는 값 구분 기호입니다. 기본값은 ','입니다.

    • "startingOffsets": (선택 사항) 데이터를 읽을 Kafka 주제의 시작 위치입니다. 가능한 값은 "earliest" 또는 "latest"입니다. 기본값은 "latest"입니다.

    • "startingTimestamp": (선택 사항, AWS Glue 버전 4.0 이상에서만 지원됨) 데이터를 읽을 Kafka 주제에 있는 레코드의 타임스탬프입니다. 가능한 값은 yyyy-mm-ddTHH:MM:SSZ 패턴에서 UTC 형식의 타임스탬프 문자열입니다(여기서, Z는 UTC 시간대 오프셋(+/-)임, 예: '2023-04-04T08:00:00-04:00').

      참고: AWS Glue 스트리밍 스크립트의 연결 옵션 목록에는 'startingOffsets' 또는 'startingTimestamp' 중 하나만 존재할 수 있습니다. 이 두 속성을 모두 포함하면 작업에 실패합니다.

    • "endingOffsets": (선택 사항) 일괄 쿼리가 종료되는 엔드포인트입니다. 가능한 값은 "latest" 또는 각 TopicPartition의 끝 오프셋을 지정하는 JSON 문자열입니다.

      JSON 문자열의 경우 포맷은 {"topicA":{"0":23,"1":-1},"topicB":{"0":-1}}입니다. 오프셋으로 값 -1은 "latest"를 나타냅니다.

    • "pollTimeoutMs": (선택 사항) Spark 작업 실행기의 Kafka에서 데이터를 폴링하는 시간 제한(밀리초)입니다. 기본값은 512입니다.

    • "numRetries": (선택 사항) Kafka 오프셋 가져오기에 실패하기 전에 재시도할 횟수입니다. 기본값은 3입니다.

    • "retryIntervalMs": (선택 사항) Kafka 오프셋을 가져오기를 다시 시도하기 전에 대기하는 시간(밀리초)입니다. 기본값은 10입니다.

    • "maxOffsetsPerTrigger": (선택 사항) 트리거 간격당 처리되는 최대 오프셋 수에 대한 속도 제한입니다. 지정된 총 오프셋 수는 서로 다른 볼륨의 topicPartitions에 비례하여 분할됩니다. 기본값은 null입니다. 즉, 소비자가 알려진 최신 오프셋까지 모든 오프셋을 읽습니다.

    • "minPartitions": (선택 사항) Kafka에서 읽을 원하는 최소 파티션 수입니다. 기본값은 null이며 이는 Spark 파티션의 수가 Kafka 파티션의 수와 동일함을 의미합니다.

    • "includeHeaders": (선택 사항) Kafka 헤더를 포함할지 여부입니다. 옵션이 "true"로 설정되면 데이터 출력에는 유형이 Array[Struct(key: String, value: String)]인 "glue_streaming_kafka_headers"라는 추가 열이 포함됩니다. 기본값은 "false"입니다. 이 옵션은 AWS Glue 버전 3.0 이상에서만 사용할 수 있습니다.

    • "schema": (InferSchema가 false로 설정된 경우 필수) 페이로드를 처리하는 데 사용할 스키마입니다. 분류가 avro인 경우 제공된 스키마는 Avro 스키마 형식이어야 합니다. 분류가 avro가 아닌 경우 제공된 스키마는 DDL 스키마 형식이어야 합니다.

      다음은 스키마의 예입니다.

      Example in DDL schema format
      'column1' INT, 'column2' STRING , 'column3' FLOAT
      Example in Avro schema format
      {
"type":"array",
"items":
{
"type":"record",
"name":"test",
"fields":
[
  {
    "name":"_id",
    "type":"string"
  },
  {
    "name":"index",
    "type":
    [
      "int",
      "string",
      "float"
    ]
  }
]
}
}

    • "inferSchema": (선택 사항) 기본값은 'false'입니다. 'true'로 설정하면 스키마가 런타임 시 foreachbatch 내의 페이로드에서 감지됩니다.

    • "avroSchema": (더 이상 사용되지 않음) Avro 형식을 사용할 때 Avro 데이터의 스키마를 지정하는 데 사용되는 파라미터입니다. 이 파라미터는 이제 사용 중단되었습니다. schema 파라미터를 사용합니다.

    • "addRecordTimestamp": (선택 사항) 이 옵션이 'true'로 설정되면 데이터 출력에는 이름이 '__src_timestamp'라는 추가 열이 포함됩니다. 이 열은 주제에서 해당 레코드를 수신한 시간을 나타냅니다. 기본값은 'false'입니다. 이 옵션은 AWS Glue 버전 4.0 이상에서 지원됩니다.

    • "emitConsumerLagMetrics": (선택 사항) 이 옵션을 'true'로 설정하면 각 배치에 대해 주제에서 수신한 가장 오래된 레코드와 AWS Glue에 도착한 시간 사이의 지표를 CloudWatch로 내보냅니다. 지표의 이름은 'glue.driver.streaming.maxConsumerLagInMs'입니다. 기본값은 'false'입니다. 이 옵션은 AWS Glue 버전 4.0 이상에서 지원됩니다.

    쓸 때는 "connectionType": "kafka"를 사용한 다음 연결 옵션을 사용합니다.

    • "connectionName"(필수) Kafka 클러스터에 연결하는 데 사용되는 AWS Glue 연결의 이름(Kafka 소스와 유사).

    • "topic"(필수) 주제 열이 존재하는 경우 주제 구성 옵션이 설정되지 않은 한 Kafka에 지정된 행을 쓸 때 해당 값이 주제로 사용됩니다. 즉, topic 구성 옵션이 주제 열을 재정의합니다.

    • "partition"(선택 사항) 유효한 파티션 번호가 지정되면 레코드를 보낼 때 해당 partition이 사용됩니다.

      파티션을 지정하지 않았지만 key가 있는 경우 키의 해시를 사용하여 파티션이 선택됩니다.

      key와 partition이 모두 존재하지 않으면 파티션에 최소 batch.size 바이트가 생성될 때 변경 사항을 고정 파티셔닝하여 파티션이 선택됩니다.

    • "key"(선택 사항) partition이 null인 경우 파티셔닝에 사용됩니다.

    • "classification"(선택 사항) 레코드의 데이터에서 사용하는 파일 형식입니다. JSON, CSV, Avro만 지원합니다.

      Avro 형식을 사용하면 직렬화할 사용자 지정 avroSchema를 제공할 수 있지만 역직렬화를 위해서는 소스에서도 이를 제공해야 한다는 점에 유의하세요. 그렇지 않으면 기본적으로 직렬화를 위해 Apache AvroSchema를 사용합니다.

    또한 필요에 따라 Kafka 프로듀서 구성 파라미터를 업데이트하여 Kafka 싱크를 미세 튜닝할 수 있습니다. 연결 옵션에는 허용 목록이 없으며 모든 키-값 쌍은 싱크에 그대로 유지됩니다.

    그러나 적용되지 않는 일부 거부 옵션의 목록이 있습니다. 자세한 내용은 Kafka 특정 구성을 참조하세요.