Amazon Kinesis - Amazon Timestream
Usando Amazon Managed Service para Apache FlinkUso de tuberías EventBridge

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Amazon Kinesis

Puede enviar datos desde Kinesis Data Streams a Timestream para utilizar el conector de datos de Timestream ejemplo LiveAnalytics para Managed Service for Apache Flink. Consulte Apache Flink Amazon Managed Service para Apache Flink para obtener más información.

Uso de EventBridge Pipes para enviar datos de Kinesis a Timestream

Puede usar EventBridge Pipes para enviar datos desde una transmisión de Kinesis a una tabla de Amazon Timestream for. LiveAnalytics

Los pipes están diseñados para la point-to-point integración entre las fuentes y los destinos compatibles, y permiten realizar transformaciones y enriquecimientos avanzados. Los tubos reducen la necesidad de conocimientos especializados y códigos de integración a la hora de desarrollar arquitecturas basadas en eventos. Para configurar una canalización, elija el origen, agregue filtros opcionales, defina el enriquecimiento opcional y elija el destino de los datos del evento.

Una fuente envía los eventos a una EventBridge canalización, que filtra y enruta los eventos coincidentes al destino.

Esta integración le permite aprovechar el potencial de las capacidades de análisis de datos Timestream de series temporales y, al mismo tiempo, simplificar el proceso de ingesta de datos.

El uso de EventBridge Pipes with Timestream ofrece las siguientes ventajas:

  • Ingestión de datos en tiempo real: transmita datos desde Kinesis directamente a Timestream LiveAnalytics para, de este modo, realizar análisis y supervisión en tiempo real.

  • Integración perfecta: utilice EventBridge Pipes para gestionar el flujo de datos sin necesidad de complejas integraciones personalizadas.

  • Filtrado y transformación mejorados: filtre o transforme los registros de Kinesis antes de almacenarlos Timestream para cumplir con sus requisitos específicos de procesamiento de datos.

  • Escalabilidad: gestione flujos de datos de alto rendimiento y garantice un procesamiento de datos eficiente con funciones integradas de paralelismo y procesamiento por lotes.

Configuración

Para configurar un EventBridge Pipe para transmitir datos de Kinesis a Timestream, siga estos pasos:

  1. Creación de un flujo de Kinesis

    Asegúrese de tener una transmisión de datos de Kinesis activa desde la que desee ingerir datos.

  2. Cree una Timestream base de datos y una tabla

    Configure la Timestream base de datos y la tabla donde se almacenarán los datos.

  3. Configure la EventBridge tubería:

    • Fuente: seleccione su transmisión de Kinesis como fuente.

    • Destino: elija Timestream como destino.

    • Configuración de procesamiento por lotes: defina la ventana de procesamiento por lotes y el tamaño del lote para optimizar el procesamiento de datos y reducir la latencia.

importante

Al configurar una canalización, recomendamos comprobar la exactitud de todas las configuraciones mediante la ingesta de algunos registros. Tenga en cuenta que la creación correcta de una canalización no garantiza que la canalización sea correcta y que los datos fluyan sin errores. Es posible que se produzcan errores de tiempo de ejecución, como una tabla incorrecta, un parámetro de ruta dinámica incorrecto o Timestream un registro no válido tras aplicar el mapeo, que se detectarán cuando los datos reales fluyan por la canalización.

Las siguientes configuraciones determinan la velocidad a la que se ingieren los datos:

  • BatchSize: El tamaño máximo del lote que se enviará a Timestream. LiveAnalytics Rango: 0 - 100. La recomendación es mantener este valor en 100 para obtener el máximo rendimiento.

  • MaximumBatchingWindowInSeconds: El tiempo máximo de espera para llenarse batchSize antes de que el lote se envíe a Timestream para su destino. LiveAnalytics En función de la velocidad de entrada de eventos, esta configuración decidirá el retraso de la ingestión. Se recomienda mantener este valor en menos de 10 segundos para seguir enviando los datos prácticamente en tiempo real. Timestream

  • ParallelizationFactor: el número de lotes de cada fragmento que se van a procesar simultáneamente. La recomendación es utilizar el valor máximo de 10 para obtener el máximo rendimiento y una ingesta casi en tiempo real.

    Si varios objetivos leen tu transmisión, utiliza una función de distribución mejorada para disponer de un consumidor dedicado a tu canal y lograr un alto rendimiento. Para obtener más información, consulta Cómo desarrollar consumidores con mayor capacidad de distribución Kinesis Data Streams API en la Guía delKinesis Data Streams usuario.

nota

El rendimiento máximo que se puede lograr está limitado por las ejecuciones simultáneas de tuberías por cuenta.

La siguiente configuración garantiza la prevención de la pérdida de datos:

  • DeadLetterConfig: Se recomienda configurarlo siempre DeadLetterConfig para evitar cualquier pérdida de datos en los casos en que los eventos no se hayan podido transferir a Timestream LiveAnalytics debido a errores del usuario.

Optimice el rendimiento de su canal con los siguientes ajustes de configuración, que ayudan a evitar que los registros se ralenticen o bloqueen.

  • MaximumRecordAgeInSeconds: Los registros más antiguos que este no se procesarán y se moverán directamente a ellos. DLQ Recomendamos establecer este valor para que no sea superior al período de retención del almacén de memoria configurado en la Timestream tabla de destino.

  • MaximumRetryAttempts: el número de reintentos de un registro antes de enviarlo a DeadLetterQueue. La recomendación es configurarlo en 10. Esto debería ayudar a solucionar cualquier problema transitorio y, en el caso de problemas persistentes, el registro se moverá al resto de la transmisión DeadLetterQueue y se desbloqueará.

  • OnPartialBatchItemFailure: En el caso de las fuentes que admiten el procesamiento parcial por lotes, le recomendamos que lo habilite y lo configure como AUTOMATIC _ BISECT para volver a intentar los registros fallidos antes de descartarlos o enviarlos a. DLQ

Ejemplo de configuración

A continuación, se muestra un ejemplo de cómo configurar un EventBridge Pipe para transmitir datos desde una transmisión de Kinesis a una Timestream tabla:

ejemplo IAM actualizaciones de políticas para Timestream
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "timestream:WriteRecords"
            ],
            "Resource": [
                "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "timestream:DescribeEndpoints"
            ],
            "Resource": "*"
        }
    ]
}
ejemplo Configuración de transmisiones de Kinesis
{
  "Source": "arn:aws:kinesis:us-east-1:123456789012:stream/my-kinesis-stream",
  "SourceParameters": {
    "KinesisStreamParameters": {
        "BatchSize": 100,
        "DeadLetterConfig": {
            "Arn": "arn:aws:sqs:us-east-1:123456789012:my-sqs-queue"
        },
       "MaximumBatchingWindowInSeconds": 5,
        "MaximumRecordAgeInSeconds": 1800,
        "MaximumRetryAttempts": 10,
        "StartingPosition": "LATEST",
       "OnPartialBatchItemFailure": "AUTOMATIC_BISECT"
    }
  }
}
ejemplo Timestream configuración de destino
{
    "Target": "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table",
    "TargetParameters": {
        "TimestreamParameters": {
            "DimensionMappings": [
                {
                    "DimensionName": "sensor_id",
                    "DimensionValue": "$.data.device_id",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_type",
                    "DimensionValue": "$.data.sensor_type",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_location",
                    "DimensionValue": "$.data.sensor_loc",
                    "DimensionValueType": "VARCHAR"
                }
            ],
            "MultiMeasureMappings": [
                {
                    "MultiMeasureName": "readings",
                    "MultiMeasureAttributeMappings": [
                        {
                            "MultiMeasureAttributeName": "temperature",
                            "MeasureValue": "$.data.temperature",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "humidity",
                            "MeasureValue": "$.data.humidity",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "pressure",
                            "MeasureValue": "$.data.pressure",
                            "MeasureValueType": "DOUBLE"
                        }
                    ]
                }
            ],
            "SingleMeasureMappings": [],
            "TimeFieldType": "TIMESTAMP_FORMAT",
            "TimestampFormat": "yyyy-MM-dd HH:mm:ss.SSS",
            "TimeValue": "$.data.time",
            "VersionValue": "$.approximateArrivalTimestamp"
        }
    }
}

Transformación de eventos

EventBridge Las tuberías le permiten transformar los datos antes de que lleguen a su destino. Timestream Puede definir reglas de transformación para modificar los Kinesis registros entrantes, como cambiar los nombres de los campos.

Suponga que su Kinesis flujo contiene datos de temperatura y humedad. Puede utilizar una EventBridge transformación para cambiar el nombre de estos campos antes de insertarlos en Timestream ellos.

Prácticas recomendadas

Almacenamiento en lotes y almacenamiento en búfer

  • Configure la ventana y el tamaño del procesamiento por lotes para equilibrar la latencia de escritura y la eficiencia del procesamiento.

  • Utilice una ventana de procesamiento por lotes para acumular suficientes datos antes de procesarlos, lo que reduce la sobrecarga de los lotes pequeños frecuentes.

Procesamiento paralelo

Utilice la ParallelizationFactorconfiguración para aumentar la simultaneidad, especialmente para transmisiones de alto rendimiento. Esto garantiza que se puedan procesar simultáneamente varios lotes de cada fragmento.

Transformación de datos

Aproveche las capacidades de transformación de EventBridge Pipes para filtrar y mejorar los registros antes de almacenarlos en ellos Timestream. Esto puede ayudar a alinear los datos con sus requisitos analíticos.

Seguridad

  • Asegúrese de que las IAM funciones utilizadas en EventBridge Pipes tengan los permisos necesarios para leer Kinesis y escribir Timestream en ellas.

  • Utilice medidas de cifrado y control de acceso para proteger los datos en tránsito y en reposo.

Fallos de depuración

  • Desactivación automática de tuberías

    Las tuberías se desactivarán automáticamente en aproximadamente 2 horas si el objetivo no existe o tiene problemas con los permisos

  • Limitaciones

    Las tuberías tienen la capacidad de retroceder automáticamente y volver a intentarlo hasta que se haya reducido el acelerador.

  • Habilitar los registros

    Le recomendamos que habilite los registros a ERROR nivel e incluya datos de ejecución para obtener más información sobre los errores. En caso de producirse un error, estos registros contendrán request/response sent/received datos de Timestream. Esto le ayuda a comprender el error asociado y, si es necesario, a volver a procesar los registros después de corregirlo.

Monitorización

Le recomendamos que configure alarmas en los siguientes lugares para detectar cualquier problema con el flujo de datos:

  • Antigüedad máxima del registro en la fuente

    • GetRecords.IteratorAgeMilliseconds

  • Métricas de fallos en tuberías

    • ExecutionFailed

    • TargetStageFailed

  • Timestream APIErrores de escritura

    • UserErrors

Para obtener métricas de supervisión adicionales, consulte Supervisión EventBridge en la Guía del EventBridge usuario.