建立頻道

使用 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. 在目的地類型下，選取 Apache Iceberg 的串流資料表。

   • 選取記錄轉換格式 (JSON 或 JSON (Glue 結構描述登錄檔）)。另請選擇純 JSON 的 Glue 結構描述登錄檔結構描述。

   • 在倉儲位置下：

     • 選取目前帳戶或跨帳戶。

     • 選取 S3 Table 儲存貯體，如果已選取跨帳戶，則輸入 S3 Table 儲存貯體 ARN。

   • 在目的地資料表下：

     • 選取或輸入命名空間。

     • 輸入資料表名稱和資料欄名稱，然後選擇壓縮類型 (ZSTD 或 Snappy)。

   • 對於 Apache Iceberg 的串流資料表，分割是以時間為基礎。

   • 設定資料新鮮度 (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。

若要為 Apache Iceberg 目的地建立具有串流資料表的頻道，請使用下列命令。

aws kafka create-channel \
    --cluster-arn "arn:aws:kafka:us-east-1:123456789012:cluster/my-express-cluster/abc123" \
    --channel-name "orders-channel" \
    --topic-configuration-list '[
        {
            "TopicArn": "arn:aws:kafka:us-east-1:123456789012:topic/my-express-cluster/abc123/orders-topic",
            "RecordConverter": { "ValueConverter": "JSON" },
            "RecordSchema": { "GsrArn": "arn:aws:glue:us-east-1:123456789012:schema/my-registry/orders-schema" }
        }
    ]' \
    --iceberg-destination-configuration '{
        "ServiceExecutionRoleArn": "arn:aws:iam::123456789012:role/MSKChannelRole",
        "DataFreshnessInSeconds": 300,
        "Catalog": {
            "WarehouseLocation": "arn:aws:s3tables:us-east-1:123456789012:bucket/analytics-table-bucket"
        },
        "DeadLetterQueueS3": {
            "BucketArn": "arn:aws:s3:::my-dlq-bucket",
            "ErrorOutputPrefix": "dlq/"
        },
        "DestinationTableList": [
            {
                "DestinationDatabaseName": "analytics_db",
                "DestinationTableName": "orders_iceberg",
                "PartitionSpec": {
                    "PartitionStrategy": "TIME_HOUR",
                    "SourceList": [ { "SourceName": "timestamp" } ]
                }
            }
        ],
        "CompressionType": "ZSTD"
    }'

• RecordConverter.ValueConverter 為 JSON（純 JSON — 提供 RecordSchema.GsrArn、定義資料的 Glue 結構描述登錄檔結構描述的 ARN) 或 JSON_SCHEMA_GSR(GSR 序列化 JSON — 結構描述 ID 內嵌在每個記錄中）。

• Catalog 是選用的。如果您提供 Catalog.CatalogArn，則必須參考 S3 Tables 目錄 (.../catalog/s3tablescatalog/<bucket-name>)；如果省略，Amazon MSK 會從 衍生它WarehouseLocation。

• CompressionType 為 ZSTD（預設） 或 SNAPPY。分割使用 PartitionStrategy: TIME_HOUR搭配單一時間戳記來源資料欄。

兩個目的地的回應都相同。

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

注意

使用 ClusterOperationArn追蹤非同步操作的狀態和任何錯誤訊息。頻道會在佈建完成ACTIVE時開始CREATING並轉換為 。

API 參考 — 請參閱《Amazon MSK API 參考CreateChannel》中的 。