Criar um canal

Usar o AWS Console de Gerenciamento

1. Abra o console do Amazon MSK em https://console.aws.amazon.com/msk/home? region=us-east-1#/home/.

2. No painel de navegação, escolha Clusters.

3. Escolha o nome do seu cluster provisionado pelo Amazon MSK com os corretores Express.

4. Escolha a guia Canal e, em seguida, escolha Criar canal.

5. Na etapa de configuração da fonte:

   • Nome do canal: insira o nome do canal a ser criado.

   • Tópico: selecione um tópico existente. Crie um novo tópico se ainda não existir.

6. Prossiga para a etapa de configuração de destino.

7. Em Tipo de destino, selecione tabelas de streaming para o Apache Iceberg.

   • Selecione um formato de conversão de registro (JSON ou JSON (Glue Schema Registry)). Escolha também o esquema Glue Schema Registry para JSON simples.

   • Em Localização do armazém:

     • Selecione Conta atual ou Conta cruzada.

     • Selecione um bucket do S3 Table ou insira um ARN do bucket do S3 Table se a conta cruzada tiver sido selecionada.

   • Na tabela de destino:

     • Selecione ou insira o namespace.

     • Insira o nome da tabela e o nome da coluna e escolha o tipo de compressão (ZSTD ou Snappy).

   • Para tabelas de streaming para o Apache Iceberg, o particionamento é baseado em tempo.

   • Defina a atualização dos dados (5 a 15 minutos).

8. Prossiga para a etapa de configuração adicional.

9. Em Configuração de criptografia, selecione opcionalmente uma chave gerenciada pelo cliente para personalizar suas configurações de criptografia.

10. Em Acesso ao serviço, selecione ou crie uma função do IAM com as permissões necessárias.

11. Em Dead-letter fila, selecione Conta atual ou Conta cruzada e especifique o URI do bucket do S3 para registros não processáveis. Opcionalmente, insira um prefixo de saída de erro. Se a opção Conta cruzada tiver sido selecionada, talvez seja necessário inserir uma ID de conta adicional.

12. (Opcional) Em Entrega de registros, selecione um grupo de CloudWatch logs da Amazon, um bucket do Amazon S3 ou um stream do Amazon Data Firehose para enviar registros para monitoramento e solução de problemas.

13. (Opcional) Adicione tags.

14. Prossiga para a etapa de revisão.

15. Revise e escolha Criar canal.

O Canal entra no estado de Criação e faz a transição para Ativo quando o pipeline de entrega é totalmente provisionado.

Usar o AWS CLI

O tópico de origem está definido em TopicConfigurationList (exatamente um tópico). O destino é definido em IcebergDestinationConfiguration ou S3DestinationConfiguration (mutuamente exclusivo), e cada configuração de destino contém sua própria ServiceExecutionRoleArn configuração DataFreshnessInSeconds (300—900, padrão 600) e. DeadLetterQueueS3

Para criar um canal com tabelas de streaming para o destino Apache Iceberg, use o comando a seguir.

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 simples — forneçaRecordSchema.GsrArn, o ARN do esquema Glue Schema Registry que define os dados) ou JSON_SCHEMA_GSR (GSR-serialized JSON — o ID do esquema está incorporado em cada registro).

• Catalog é opcional. Se você fornecerCatalog.CatalogArn, ele deverá fazer referência ao catálogo de tabelas do S3 (.../catalog/s3tablescatalog/<bucket-name>); se omitido, o Amazon MSK o derivará de. WarehouseLocation

• CompressionTypeé ZSTD (padrão) ouSNAPPY. O particionamento é usado PartitionStrategy: TIME_HOUR com uma única coluna de origem de carimbo de data/hora.

A resposta é a mesma para os dois destinos.

{
    "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

Use o ClusterOperationArn para rastrear o status da operação assíncrona e qualquer mensagem de erro. O canal começa CREATING e passa para ACTIVE quando o provisionamento é concluído.

Referência de API — veja CreateChannel na Referência de API do Amazon MSK.