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. 送信先タイプで、Apache Iceberg のストリーミングテーブルを選択します。

    • レコード変換形式 (JSON または JSON (Glue Schema Registry)) を選択します。また、プレーン JSON の Glue スキーマレジストリスキーマを選択します。

    • 倉庫の場所:

      • 現在のアカウントまたはクロスアカウントを選択します。

      • S3 テーブルバケットを選択するか、クロスアカウントが選択されている場合は S3 テーブルバケット ARN を入力します。

    • 送信先テーブルの下:

      • 名前空間を選択または入力します。

      • テーブル名と列名を入力し、圧縮タイプ (ZSTD または Snappy) を選択します。

    • Apache Iceberg のストリーミングテーブルの場合、パーティション分割は時間ベースです。

    • データの鮮度 (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。

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 — データを定義する Glue スキーマレジストリスキーマの RecordSchema.GsrArnARN を提供) または 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、非同期オペレーションのステータスとエラーメッセージを追跡します。チャネルは で開始CREATINGされ、ACTIVEプロビジョニングが完了すると に移行します。

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