Creare un canale

Utilizzo di AWS Console di gestione

1. Apri la console Amazon MSK all'indirizzo https://console.aws.amazon.com/msk/home? region=us-east-1#/home/.

2. Nel pannello di navigazione scegliere Cluster.

3. Scegli il nome del tuo cluster Amazon MSK Provisioned con i broker Express.

4. Scegli la scheda Canale, quindi scegli Crea canale.

5. Nella fase di configurazione del codice sorgente:

   • Nome del canale: inserisci il nome del canale da creare.

   • Argomento: seleziona un argomento esistente. Crea un nuovo argomento se non ne esiste ancora uno.

6. Procedi alla fase di configurazione della destinazione.

7. In Tipo di destinazione, seleziona le tabelle di streaming per Apache Iceberg.

   • Seleziona un formato di conversione dei record (JSON o JSON (Glue Schema Registry)). Scegliete anche lo schema Glue Schema Registry per un semplice JSON.

   • In Ubicazione del magazzino:

     • Seleziona Conto corrente o Conto incrociato.

     • Seleziona un bucket S3 Table o inserisci un ARN del bucket S3 Table se è stato selezionato Cross account.

   • In Tabella di destinazione:

     • Seleziona o inserisci il Namespace.

     • Inserite il nome della tabella e il nome della colonna e scegliete il tipo di compressione (ZSTD o Snappy).

   • Per le tabelle di streaming per Apache Iceberg, il partizionamento è basato sul tempo.

   • Imposta l'aggiornamento dei dati (5-15 minuti).

8. Procedi alla fase di configurazione aggiuntiva.

9. In Configurazione della crittografia, seleziona facoltativamente una chiave gestita dal cliente per personalizzare le impostazioni di crittografia.

10. In Service Access, seleziona o crea un ruolo IAM con le autorizzazioni richieste.

11. In Dead-letter coda, seleziona Account corrente o Account incrociato e specifica l'URI del bucket S3 per i record non elaborabili. Facoltativamente, inserisci un prefisso di output dell'errore. Se è stato selezionato Cross account, potrebbe essere necessario inserire un ID account aggiuntivo.

12. (Facoltativo) In Log Delivery, seleziona un gruppo di CloudWatch log Amazon, un bucket Amazon S3 o uno stream Amazon Data Firehose per inviare i log per il monitoraggio e la risoluzione dei problemi.

13. (Facoltativo) Aggiungi tag.

14. Procedi alla fase di revisione.

15. Rivedi e scegli Crea canale.

Il canale passa allo stato Creazione e passa ad Attivo una volta che la pipeline di distribuzione è stata completamente predisposta.

Utilizzo di AWS CLI

L'argomento di origine è impostato in TopicConfigurationList (esattamente un argomento). La destinazione è impostata in IcebergDestinationConfiguration o S3DestinationConfiguration (si escludono a vicenda) e ogni configurazione di destinazione contiene i propriServiceExecutionRoleArn, DataFreshnessInSeconds (300-900, impostazione predefinita 600) e. DeadLetterQueueS3

Per creare un canale con tabelle di streaming per la destinazione Apache Iceberg, usa il seguente comando.

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 semplice: fornisce RecordSchema.GsrArn l'ARN dello schema Glue Schema Registry che definisce i dati) o JSON_SCHEMA_GSR (GSR-serialized JSON: l'ID dello schema è incorporato in ogni record).

• Catalog è facoltativo. Se lo fornisciCatalog.CatalogArn, deve fare riferimento al catalogo S3 Tables (.../catalog/s3tablescatalog/<bucket-name>); se omesso, Amazon MSK lo deriva da. WarehouseLocation

• CompressionTypeè ZSTD (impostazione predefinita) o. SNAPPY Il partizionamento viene utilizzato PartitionStrategy: TIME_HOUR con una singola colonna di origine del timestamp.

La risposta è la stessa per entrambe le destinazioni.

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

Nota

Usa il ClusterOperationArn per tenere traccia dello stato dell'operazione asincrona e di eventuali messaggi di errore. Il canale inizia CREATING e passa al termine del provisioning. ACTIVE

Riferimento API: consulta la sezione Amazon CreateChannel MSK API Reference.