View a markdown version of this page

建立頻道 - Amazon Managed Streaming for Apache Kafka
使用 AWS 管理主控台使用 AWS CLI

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

建立頻道

使用 AWS 管理主控台

  1. 開啟 Amazon MSK 主控台，網址為 https://console.aws.amazon.com/msk/home?region=us-east-1#/home/。

  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 一般用途儲存貯體， RecordConverter.ValueConverter 可以是 JSON、 BYTE_ARRAY或 STRING（不需要結構描述登錄）。在 中Storage， CompressionType可以是 NONE、 GZIP或 ZSTD，而 StorageClass可以是 INTELLIGENT_TIERING、 STANDARD或 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 參考 — 請參閱《Amazon MSK API 參考CreateChannel》中的 。