Creación de un canal

Uso de AWS Consola de administración

1. Abra la consola de Amazon MSK en? https://console.aws.amazon.com/msk/home region=us-east-1#/home/.

2. En el panel de navegación, seleccione Clusters (Clústeres).

3. Elija el nombre de su clúster aprovisionado de Amazon MSK con los agentes de Express.

4. Seleccione la pestaña Canal y, a continuación, elija Crear canal.

5. En el paso de configuración de la fuente:

   • Nombre del canal: introduzca el nombre del canal que desea crear.

   • Tema: selecciona un tema existente. Crea un tema nuevo si aún no existe.

6. Continúe con el paso de configuración del destino.

7. En Tipo de destino, selecciona las tablas de streaming para Apache Iceberg.

   • Selecciona un formato de conversión de registros (JSON o JSON (Glue Schema Registry)). Elija también el esquema de registro de Glue Schema para JSON simple.

   • En Ubicación del almacén:

     • Selecciona Cuenta corriente o Cuenta cruzada.

     • Selecciona un bucket de S3 Table o introduce un ARN de S3 Table bucket si has seleccionado una cuenta de Cross.

   • En la tabla de destinos:

     • Seleccione o introduzca el espacio de nombres.

     • Introduzca el nombre de la tabla y el nombre de la columna y elija el tipo de compresión (ZSTD o Snappy).

   • En el caso de las tablas de streaming de Apache Iceberg, la partición se basa en el tiempo.

   • Establezca la actualización de los datos (de 5 a 15 minutos).

8. Continúe con el paso de configuración adicional.

9. En Configuración de cifrado, si lo desea, seleccione una clave gestionada por el cliente para personalizar la configuración de cifrado.

10. En Acceso al servicio, selecciona o crea un rol de IAM con los permisos necesarios.

11. En Dead-letter cola, selecciona Cuenta actual o Cuenta cruzada y especifica el URI del bucket de S3 para los registros no procesables. Si lo desea, introduzca un prefijo de salida de error. Si se seleccionó la cuenta cruzada, es posible que tengas que introducir un identificador de cuenta adicional.

12. (Opcional) En Entrega de registros, selecciona un grupo de CloudWatch registros de Amazon, un bucket de Amazon S3 o una transmisión de Amazon Data Firehose para enviar los registros para su supervisión y resolución de problemas.

13. (Opcional) Agregue etiquetas.

14. Continúe con el paso de revisión.

15. Revisa y elige Crear canal.

El canal pasa al estado de creación y pasa a estar activo una vez que la canalización de entrega esté completamente aprovisionada.

Uso de AWS CLI

El tema de origen se establece en TopicConfigurationList (exactamente un tema). El destino se establece en IcebergDestinationConfiguration o S3DestinationConfiguration (excluyentes entre sí) y cada configuración de destino contiene la suya propia ServiceExecutionRoleArn DataFreshnessInSeconds (300—900, por defecto 600) y. DeadLetterQueueS3

Para crear un canal con tablas de streaming para el destino de Apache Iceberg, utilice el siguiente 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.ValueConverteres JSON (JSON simple: proporcione RecordSchema.GsrArn el ARN del esquema de registro de Glue que define los datos) o JSON_SCHEMA_GSR (GSR-serialized JSON: el ID del esquema está incrustado en cada registro).

• Catalog es opcional. Si lo proporcionaCatalog.CatalogArn, debe hacer referencia al catálogo de tablas de S3 (.../catalog/s3tablescatalog/<bucket-name>); si se omite, Amazon MSK lo deriva. WarehouseLocation

• CompressionTypees ZSTD (predeterminado) o. SNAPPY La partición se utiliza PartitionStrategy: TIME_HOUR con una sola columna fuente con marca de tiempo.

La respuesta es la misma para ambos 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

Utilice el ClusterOperationArn para realizar un seguimiento del estado de la operación asincrónica y de cualquier mensaje de error. El canal comienza CREATING y pasa a ACTIVE cuando finaliza el aprovisionamiento.

Referencia de la API: consulte CreateChannel la referencia de la API de Amazon MSK.