View a markdown version of this page

채널 생성 - Amazon Managed Streaming for Apache Kafka
AWS 관리 콘솔 사용AWS CLI 사용

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

채널 생성

AWS 관리 콘솔 사용

  1. https://console.aws.amazon.com/msk/home?region=us-east-1#/home/에서 Amazon MSK 콘솔을 엽니다.

  2. 탐색 창에서 클러스터를 선택합니다.

  3. Express 브로커가 있는 Amazon MSK 프로비저닝된 클러스터의 이름을 선택합니다.

  4. 채널 탭을 선택한 다음 채널 생성을 선택합니다.

  5. 소스 구성 단계에서:

    • 채널 이름: 생성할 채널의 이름을 입력합니다.

    • 주제: 기존 주제를 선택합니다. 아직 존재하지 않는 경우 새 주제를 생성합니다.

  6. 대상 구성 단계로 이동합니다.

  7. 대상 유형에서 범용 S3로 데이터 전송을 선택합니다.

    • 레코드 형식(JSON, 바이트 배열 또는 문자열)을 선택합니다.

    • 대상 설정에서:

      • 현재 계정 또는 교차 계정을 선택합니다. 교차 계정을 선택한 경우 추가 계정 ID를 입력해야 할 수 있습니다.

      • 전송 버킷에 대한 S3 버킷 URI를 선택하거나 입력합니다.

      • 선택적 S3 버킷 접두사 및 출력 키 템플릿을 입력합니다.

      • 스토리지 클래스 및 압축 유형(NONE, GZIP 또는 ZSTD)을 선택합니다.

    • 데이터 신선도(5~15분)를 설정합니다.

  8. 추가 구성 단계로 이동합니다.

  9. 암호화 구성에서 선택적으로 고객 관리형 키를 선택하여 암호화 설정을 사용자 지정합니다.

  10. 서비스 액세스에서 필요한 권한이 있는 IAM 역할을 선택하거나 생성합니다.

  11. 배달 못한 편지 대기열에서 현재 계정 또는 교차 계정을 선택하고 처리할 수 없는 레코드에 대한 S3 버킷 URI를 지정합니다. 선택적으로 오류 출력 접두사를 입력합니다. 교차 계정을 선택한 경우 추가 계정 ID를 입력해야 할 수 있습니다.

  12. (선택 사항) 로그 전송에서 Amazon CloudWatch 로그 그룹, Amazon S3 버킷 또는 Amazon Data Firehose 스트림을 선택하여 모니터링 및 문제 해결을 위한 로그를 전송합니다.

  13. (선택 사항) 태그를 추가합니다.

  14. 검토 단계로 이동합니다.

  15. 채널 생성을 검토하고 선택합니다.

전송 파이프라인이 완전히 프로비저닝되면 채널이 생성 중 상태로 전환되고 활성으로 전환됩니다.

AWS CLI 사용

소스 주제는 TopicConfigurationList (정확히 하나의 주제)에 설정되어 있습니다. 대상은 IcebergDestinationConfiguration 또는 S3DestinationConfiguration (상호 제외)로 설정되며, 각 대상 구성에는 자체 ServiceExecutionRoleArn, DataFreshnessInSeconds (300~900, 기본값 600) 및가 포함됩니다DeadLetterQueueS3.

Amazon S3 범용 버킷 대상으로 채널을 생성하려면 다음 명령을 사용합니다.

aws kafka create-channel \
    --cluster-arn "arn:aws:kafka:us-east-1:123456789012:cluster/my-express-cluster/abc123" \
    --channel-name "orders-s3-channel" \
    --topic-configuration-list '[
        {
            "TopicArn": "arn:aws:kafka:us-east-1:123456789012:topic/my-express-cluster/abc123/orders-topic",
            "RecordConverter": { "ValueConverter": "JSON" }
        }
    ]' \
    --s3-destination-configuration '{
        "ServiceExecutionRoleArn": "arn:aws:iam::123456789012:role/MSKChannelRole",
        "DataFreshnessInSeconds": 300,
        "DeadLetterQueueS3": {
            "BucketArn": "arn:aws:s3:::my-dlq-bucket",
            "ErrorOutputPrefix": "dlq/"
        },
        "Storage": {
            "BucketArn": "arn:aws:s3:::my-delivery-bucket",
            "OutputPrefix": "expresslink/",
            "OutputKeyTemplate": "!{channel-id}/!{topic-name}/year=!{yyyy}/month=!{MM}/day=!{dd}/hour=!{HH}/!{topic-name}+!{partition-id}+!{kafka-offset}",
            "StorageClass": "STANDARD",
            "CompressionType": "GZIP"
        }
    }'

Amazon S3 범용 버킷의 경우는 JSON, BYTE_ARRAY또는 일 RecordConverter.ValueConverter 수 있습니다STRING(스키마 레지스트리 필요 없음). 에서는 NONE, GZIP또는 일 Storage CompressionType 수 있으며 ZSTD, STANDARD INTELLIGENT_TIERING또는 일 StorageClass 수 있습니다GLACIER_IR.

응답은 두 대상에서 동일합니다.

{
    "ChannelArn": "arn:aws:kafka:us-east-1:123456789012:channel/my-express-cluster/abc123/orders-s3-channel",
    "ClusterOperationArn": "arn:aws:kafka:us-east-1:123456789012:cluster-operation/my-express-cluster/abc123/..."
}
참고

ClusterOperationArn를 사용하여 비동기 작업의 상태와 오류 메시지를 추적합니다. 채널은에서 시작CREATING되며 ACTIVE 프로비저닝이 완료되면 로 전환됩니다.

API 참조 - Amazon MSK API 참조CreateChannel의 섹션을 참조하세요.