View a markdown version of this page

创建通道 - Amazon Managed Streaming for Apache Kafka
使用 AWS 管理控制台使用 AWS CLI

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

创建通道

使用 AWS 管理控制台

  1. 在？处打开亚马逊 MSK 控制台 https://console.aws.amazon.com/msk/homeregion=us -east-1#/home/。

  2. 在导航窗格中，选择集群。

  3. 选择您的带有 Express 代理的亚马逊 MSK 预置集群的名称。

  4. 选择 “频道” 选项卡，然后选取 “创建频道” 。

  5. 在源配置步骤中：

    • 频道名称：输入要创建的频道的名称。

    • 主题：选择现有主题。如果尚不存在，请创建一个新主题。

  6. 继续执行目标配置步骤。

  7. 在 “目标类型” 下，选择向通用型 S3 传输数据。

    • 选择记录格式（JSON、字节数组或字符串）。

    • 在 “目的地设置” 下：

      • 选择 “当前账户” 或 “交叉账户” 。如果选择了交叉账户，则可能需要输入额外的账户 ID 。

      • 为交付存储桶选择或输入 S3 存储桶 URI。

      • 输入可选的 S3 存储桶前缀和输出密钥模板。

      • 选择存储类和压缩类型（无、GZIP 或 ZSTD）。

    • 设置数据新鲜度（5—15 分钟）。

  8. 继续执行其他配置步骤。

  9. 在 “加密配置” 下，可以选择一个客户管理的密钥来自定义您的加密设置。

  10. 在 “服务访问权限” 下，选择或创建具有所需权限的 IAM 角色。

  11. 在 “Dead-letter 队列” 下，选择 “当前账户” 或 “跨账户”，然后为不可处理的记录指定 S3 存储桶 URI。（可选）输入错误输出前缀。如果选择了交叉账户，则可能需要输入额外的账户 ID 。

  12. （可选）在 “日志传输” 下，选择亚马逊 CloudWatch 日志组、Amazon S3 存储桶或 Amazon Data Firehose 流，以发送用于监控和故障排除的日志。

  13. （可选）添加标签。

  14. 继续执行审阅步骤。

  15. 查看并选择创建频道。

完全配置交付管道后，频道进入创建状态并过渡到活动。

使用 AWS CLI

源主题设置在TopicConfigurationList（正好一个主题）中。目标设置为IcebergDestinationConfiguration或S3DestinationConfiguration（互斥），并且每个目标配置都包含自己的DataFreshnessInSeconds（300—900ServiceExecutionRoleArn，默认 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，RecordConverter.ValueConverter可以是BYTE_ARRAY、或STRING（无需架构注册表）。在Storage、CompressionType可能是NONEGZIP、或ZSTD，也StorageClass可能是STANDARDINTELLIGENT_TIERING、或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该频道在配置完成ACTIVE时开始CREATING并过渡到。

API 参考 — 参见CreateChannel亚马逊 MSK API 参考文档。