Buat Saluran

Menggunakan AWS Konsol Manajemen

1. Buka konsol Amazon MSK di? https://console.aws.amazon.com/msk/home wilayah = us-east-1#/rumah/.

2. Pada panel navigasi, silakan pilih Klaster.

3. Pilih nama cluster Amazon MSK Provisioned Anda dengan broker Express.

4. Pilih tab Saluran, lalu pilih Buat Saluran.

5. Pada langkah konfigurasi Sumber:

   • Nama saluran: masukkan nama Saluran yang akan dibuat.

   • Topik: pilih topik yang ada. Buat topik baru jika belum ada.

6. Lanjutkan ke langkah konfigurasi tujuan.

7. Di bawah Tipe tujuan, pilih tabel streaming untuk Apache Iceberg.

   • Pilih format konversi Rekaman (JSON atau JSON (Glue Schema Registry)). Juga pilih skema Glue Schema Registry untuk JSON biasa.

   • Di bawah lokasi Gudang:

     • Pilih Akun saat ini atau Akun silang.

     • Pilih bucket Tabel S3, atau masukkan bucket Tabel S3 ARN jika akun Lint as dipilih.

   • Di bawah tabel Tujuan:

     • Pilih atau masukkan Namespace.

     • Masukkan nama Tabel dan nama Kolom, dan pilih jenis Kompresi (ZSTD atau Snappy).

   • Untuk tabel streaming untuk Apache Iceberg, Partisi berbasis waktu.

   • Atur kesegaran data (5—15 menit).

8. Lanjutkan ke langkah konfigurasi tambahan.

9. Di bawah Konfigurasi enkripsi, pilih kunci yang dikelola pelanggan secara opsional untuk menyesuaikan pengaturan enkripsi Anda.

10. Di bawah Ak ses Layanan, pilih atau buat peran IAM dengan izin yang diperlukan.

11. Di bawah Dead-letter antrian, pilih Akun saat ini atau Akun silang dan tentukan URI bucket S3 untuk catatan yang tidak dapat diproses. Secara opsional masukkan awalan keluaran kesalahan. Jika Cross account dipilih, Anda mungkin perlu memasukkan ID Akun tambahan.

12. (Opsional) Di bawah Pengiriman Log, pilih grup CloudWatch log Amazon, bucket Amazon S3, atau aliran Amazon Data Firehose untuk mengirim log untuk pemantauan dan pemecahan masalah.

13. (Opsional) Tambahkan tanda.

14. Lanjutkan ke langkah Review.

15. Tinjau dan pilih Buat Saluran.

Saluran memasuki status Membuat dan beralih ke Aktif setelah pipeline pengiriman sepenuhnya disediakan.

Menggunakan AWS CLI

Topik sumber diatur dalam TopicConfigurationList (tepat satu topik). Tujuan diatur dalam salah satu IcebergDestinationConfiguration atau S3DestinationConfiguration (saling eksklusif), dan setiap konfigurasi tujuan berisi konfigurasi sendiriServiceExecutionRoleArn, DataFreshnessInSeconds (300—900, default 600), dan. DeadLetterQueueS3

Untuk membuat Saluran dengan tabel streaming untuk tujuan Apache Iceberg, gunakan perintah berikut.

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.ValueConverteradalah JSON (JSON biasa — berikanRecordSchema.GsrArn, ARN dari skema Glue Schema Registry yang mendefinisikan data) atau JSON_SCHEMA_GSR (GSR-serialized JSON — ID skema disematkan di setiap catatan).

• Catalogbersifat opsional. Jika Anda memberikanCatalog.CatalogArn, itu harus mereferensikan katalog Tabel S3 (.../catalog/s3tablescatalog/<bucket-name>); jika dihilangkan, Amazon MSK mendapatkannya dari. WarehouseLocation

• CompressionTypeadalah ZSTD (default) atauSNAPPY. Partisi digunakan PartitionStrategy: TIME_HOUR dengan kolom sumber stempel waktu tunggal.

Jawabannya sama untuk kedua tujuan.

{
    "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/..."
}

catatan

Gunakan ClusterOperationArn untuk melacak status operasi asinkron dan pesan kesalahan apa pun. Saluran mulai masuk CREATING dan beralih ke ACTIVE saat penyediaan selesai.

Referensi API — lihat CreateChannel di Refer ensi API Amazon MSK.