View a markdown version of this page

チャネルの作成 - Amazon Managed Streaming for Apache Kafka
AWS マネジメントコンソールの使用CLI AWS の使用

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

チャネルの作成

AWS マネジメントコンソールの使用

  1. https://console.aws.amazon.com/msk/home?region=us-east-1#/home/ で Amazon MSK コンソールを開きます。

  2. ナビゲーションペインで [Clusters] (クラスター) を選択してください。

  3. Express ブローカーで Amazon MSK プロビジョンドクラスターの名前を選択します。

  4. Channel タブを選択し、Create Channel を選択します。

  5. ソース設定ステップ:

    • チャネル名: 作成するチャネルの名前を入力します。

    • トピック: 既存のトピックを選択します。まだ存在しない場合は、新しいトピックを作成します。

  6. 送信先設定ステップに進みます。

  7. 送信先タイプで、汎用 S3 へのデータ配信を選択します。

    • レコード形式 (JSON、バイト配列、または文字列) を選択します。

    • 送信先設定:

      • 現在のアカウントまたはクロスアカウントを選択します。クロスアカウントを選択した場合は、追加のアカウント ID の入力が必要になる場合があります。

      • 配信バケットの S3 バケット URI を選択または入力します。

      • オプションの S3 バケットプレフィックスと出力キーテンプレートを入力します。

      • ストレージクラスと圧縮タイプ (NONE、GZIP、または ZSTD) を選択します。

    • データの鮮度 (5～15 分) を設定します。

  8. 追加設定ステップに進みます。

  9. 暗号化設定で、オプションでカスタマーマネージドキーを選択して暗号化設定をカスタマイズします。

  10. Service Access で、必要なアクセス許可を持つ IAM ロールを選択または作成します。

  11. デッドレターキューで、現在のアカウントまたはクロスアカウントを選択し、処理不可能なレコードの S3 バケット URI を指定します。必要に応じて、エラー出力プレフィックスを入力します。クロスアカウントを選択した場合は、追加のアカウント ID の入力が必要になる場合があります。

  12. (オプション) ログ配信で、Amazon CloudWatch ロググループ、Amazon S3 バケット、または Amazon Data Firehose ストリームを選択して、モニタリングとトラブルシューティングのためにログを送信します。

  13. (オプション) タグを追加します。

  14. レビューステップに進みます。

  15. チャネルの作成を確認して選択します。

チャネルは作成状態になり、配信パイプラインが完全にプロビジョニングされるとアクティブに移行します。

CLI AWS の使用

ソーストピックは TopicConfigurationList (正確に 1 つのトピック) で設定されます。送信先は 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 (スキーマレジストリは必要ありません）。ではStorage、 CompressionType は NONE、GZIP、または でZSTD、 StorageClassは STANDARD、INTELLIGENT_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、非同期オペレーションのステータスとエラーメッセージを追跡します。チャネルは で始まりCREATING、ACTIVEプロビジョニングが完了すると に移行します。

API リファレンス — Amazon MSK API リファレンスのCreateChannel「」を参照してください。