Consultas de puesta al día programadas Ejecuciones manuales para datos impredecibles que llegan tarde

Gestión de datos que llegan tarde

Es posible que haya situaciones en las que los datos lleguen con bastante retraso; por ejemplo, el momento en que se ingresaron los datos en Timestream para LiveAnalytics se retrasa de manera considerable en comparación con la marca de tiempo asociada a las filas que se ingieren. En los ejemplos anteriores, hemos visto cómo puede utilizar los intervalos de tiempo definidos por el parámetro @scheduled_runtime para tener en cuenta algunos datos que llegan tarde. Sin embargo, si tiene casos de uso en los que los datos pueden retrasarse horas o días, es posible que necesite un patrón diferente para asegurarse de que los cálculos previos de la tabla derivada se actualizan de manera adecuada para reflejar los datos que llegan tarde. Para obtener información general sobre los datos que llegan tarde, consulte Datos de escritura (inserciones y ajustes).

A continuación, encontrará dos formas distintas de abordar estos datos que llegan tarde.

Si tiene retrasos predecibles en la llegada de los datos, puede utilizar otro cálculo programado para ponerse al día a fin de actualizar los agregados en función de los datos que lleguen tarde.
Si tiene retrasos impredecibles o, en algunas ocasiones los datos llegan tarde, puede utilizar las ejecuciones manuales para actualizar las tablas derivadas.

En este análisis se describen las situaciones de llegada tardía de los datos. Sin embargo, los mismos principios se aplican a las correcciones de datos, en las que se modifican los datos de la tabla de origen y se quieren actualizar los agregados de las tablas derivadas.

Temas

Consultas de puesta al día programadas
Ejecuciones manuales para datos impredecibles que llegan tarde

Consultas de puesta al día programadas

Consulta agregando datos que llegaron a tiempo

A continuación, encontrará un patrón en el que se muestra cómo se pueden utilizar una manera automática para actualizar los agregados en caso de que se produzcan retrasos predecibles en la llegada de los datos. Considere uno de los ejemplos anteriores de un cálculo programado con datos en tiempo real que se muestra a continuación. Este cálculo programado actualiza la tabla derivada una vez cada 30 minutos y ya tiene en cuenta los datos con un retraso de hasta una hora.


{
    "Name": "MultiPT30mPerHrPerTimeseriesDPCount",
    "QueryString": "SELECT region, cell, silo, availability_zone, microservice_name, instance_type, os_version, instance_name, process_name, jdk_version, bin(time, 1h) as hour, SUM(CASE WHEN measure_name = 'metrics' THEN 20 ELSE 5 END) as numDataPoints FROM raw_data.devops WHERE time BETWEEN bin(@scheduled_runtime, 1h) - 1h AND @scheduled_runtime + 1h GROUP BY region, cell, silo, availability_zone, microservice_name, instance_type, os_version, instance_name, process_name, jdk_version, bin(time, 1h)",
    "ScheduleConfiguration": {
        "ScheduleExpression": "cron(0/30 * * * ? *)"
    },
    "NotificationConfiguration": {
        "SnsConfiguration": {
            "TopicArn": "******"
        }
    },
    "TargetConfiguration": {
        "TimestreamConfiguration": {
            "DatabaseName": "derived",
            "TableName": "dp_per_timeseries_per_hr",
            "TimeColumn": "hour",
            "DimensionMappings": [
                {
                    "Name": "region",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "cell",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "silo",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "availability_zone",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "microservice_name",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "instance_type",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "os_version",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "instance_name",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "process_name",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "jdk_version",
                    "DimensionValueType": "VARCHAR"
                }
            ],
            "MultiMeasureMappings": {
                "TargetMultiMeasureName": "numDataPoints",
                "MultiMeasureAttributeMappings": [
                    {
                        "SourceColumn": "numDataPoints",
                        "MeasureValueType": "BIGINT"
                    }
                ]
            }
        }
    },
    "ErrorReportConfiguration": {
        "S3Configuration" : {
            "BucketName" : "******",
            "ObjectKeyPrefix": "errors",
            "EncryptionOption": "SSE_S3"
        }
    },
    "ScheduledQueryExecutionRoleArn": "******"
}

Consulte de puesta al día que actualiza los agregados de los datos que llegan tarde

Si tiene en cuenta el caso de que los datos pueden retrasarse unas 12 horas. A continuación, se muestra una variante de la misma consulta. Sin embargo, la diferencia es que calcula los agregados a partir de los datos que se retrasan hasta 12 horas en comparación con el momento en que se activa el cálculo programado. Por ejemplo, si ve la consulta en el siguiente ejemplo, el intervalo de tiempo al que se dirige esta consulta es entre 2 y 14 horas antes de que se la active. Además, si observa la expresión de programación cron(0 0,12 * * ? *), activará el cálculo a las 00:00 UTC y a las 12:00 UTC todos los días. Por lo tanto, cuando la consulta se active el 01-12-2021 00:00:00, la consulta actualizará los agregados en el intervalo de tiempo entre 30-11-2021 10:00:00 y 30-11-2021 22:00:00. Las consultas programadas utilizan una semántica ascendente similar a la de las escrituras de Timestream para LiveAnalytics, donde esta consulta de ponerse al día actualizará los valores agregados con valores más nuevos si existen datos que llegan tarde a la ventana o si se encuentran agregados más nuevos (por ejemplo, aparece una nueva agrupación en este agregado que no estaba presente cuando se activó el cálculo programado original), luego el nuevo agregado se insertará en la tabla derivada. Del mismo modo, cuando la siguiente instancia se active el 01-12-2021 12:00:00, esa instancia actualizará los agregados en el rango de 30-11-2021 22:00:00 a 01-12-2021 10:00:00.



       {
    "Name": "MultiPT12HPerHrPerTimeseriesDPCountCatchUp",
    "QueryString": "SELECT region, cell, silo, availability_zone, microservice_name, instance_type, os_version, instance_name, process_name, jdk_version, bin(time, 1h) as hour, SUM(CASE WHEN measure_name = 'metrics' THEN 20 ELSE 5 END) as numDataPoints FROM raw_data.devops WHERE time BETWEEN bin(@scheduled_runtime, 1h) - 14h AND bin(@scheduled_runtime, 1h) - 2h GROUP BY region, cell, silo, availability_zone, microservice_name, instance_type, os_version, instance_name, process_name, jdk_version, bin(time, 1h)",
    "ScheduleConfiguration": {
        "ScheduleExpression": "cron(0 0,12 * * ? *)"
    },
    "NotificationConfiguration": {
        "SnsConfiguration": {
            "TopicArn": "******"
        }
    },
    "TargetConfiguration": {
        "TimestreamConfiguration": {
            "DatabaseName": "derived",
            "TableName": "dp_per_timeseries_per_hr",
            "TimeColumn": "hour",
            "DimensionMappings": [
                {
                    "Name": "region",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "cell",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "silo",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "availability_zone",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "microservice_name",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "instance_type",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "os_version",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "instance_name",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "process_name",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "Name": "jdk_version",
                    "DimensionValueType": "VARCHAR"
                }
            ],
            "MultiMeasureMappings": {
                "TargetMultiMeasureName": "numDataPoints",
                "MultiMeasureAttributeMappings": [
                    {
                        "SourceColumn": "numDataPoints",
                        "MeasureValueType": "BIGINT"
                    }
                ]
            }
        }
    },
    "ErrorReportConfiguration": {
        "S3Configuration" : {
            "BucketName" : "******",
            "ObjectKeyPrefix": "errors",
            "EncryptionOption": "SSE_S3"
        }
    },
    "ScheduledQueryExecutionRoleArn": "******"
}

El ejemplo anterior es uno en el que se supone que su llegada tardía está limitada a 12 horas y que está bien actualizar la tabla derivada una vez cada 12 horas para que los datos lleguen más tarde del intervalo de tiempo real. Puede adaptar este patrón para actualizar la tabla derivada una vez por hora, de modo que la tabla derivada refleje antes los datos que llegan tarde. Del mismo modo, puede adaptar el intervalo de tiempo para que tenga más de 12 horas, por ejemplo, un día o incluso una semana o más, a fin de gestionar datos predecibles que lleguen tarde.

Ejecuciones manuales para datos impredecibles que llegan tarde

Puede haber casos en los que tenga datos impredecibles que lleguen tarde o en los que haya realizado cambios en los datos de origen y actualizado algunos valores posteriormente. En todos estos casos, puede activar de forma manual las consultas programadas para actualizar la tabla derivada. A continuación, se muestra un ejemplo de cómo puede realizarlo.

Suponga que tiene un caso de uso en el que el cálculo está escrito en la tabla derivada dp_per_timeseries_per_hr. Los datos base de la tabla DevOps se actualizaron en el intervalo de tiempo entre 30-11-2021 23:00:00 y 01-12-2021 00:00:00. Existen dos consultas programadas diferentes que se pueden usar para actualizar esta tabla derivada: MultiPT30mPerHrPerTimeseriesDPCount y MultiPT12HPerHrPerTimeseriesDPCountCatchUp. Cada cálculo programado que cree en Timestream para LiveAnalytics tiene un ARN único que se obtiene al crear el cálculo o realizar una operación de lista. Puede usar el ARN para el cálculo y un valor para el parámetro @scheduled_runtime utilizado por la consulta a fin de realizar esta operación.

Suponga que el cálculo de MultiPT30mPerHrPerTimeseriesDPCount tiene un ARN arn_1 y quiere utilizar este cálculo para actualizar la tabla derivada. Dado que el cálculo programado anterior actualiza los agregados 1 hora antes y 1 hora después del valor @scheduled_runtime, puede cubrir el intervalo de tiempo de la actualización (entre 30-11-2021 23:00:00 y 01-12-2021 00:00:00) mediante el uso del valor 01-12-2021 00:00:00 para el parámetro @scheduled_runtime. Para ello, puede utilizar la API ExecuteScheduledQuery para pasar el ARN de este cálculo y el valor del parámetro de tiempo en segundos de época (en UTC). A continuación, se muestra un ejemplo con la CLI de AWS y puede seguir el mismo patrón con cualquiera de los SDK compatibles con Timestream para LiveAnalytics.


aws timestream-query execute-scheduled-query --scheduled-query-arn arn_1 --invocation-time 1638316800 --profile profile --region us-east-1

En el ejemplo anterior, el perfil es el perfil de AWS que tiene los privilegios adecuados para realizar esta llamada a la API y 1638316800 corresponde a la segunda época del 01-12-2021 00:00:00. Este activador manual se comporta casi igual que el activador automático, suponiendo que el sistema active esta invocación en el periodo de tiempo deseado.

Si realizó una actualización durante un periodo de tiempo más largo, supongamos que los datos base se actualizaron entre 30-11-2021 23:00:00 y 01-12-2021 11:00:00, puede activar las consultas anteriores varias veces para cubrir todo este intervalo de tiempo. Por ejemplo, puede realizar seis ejecuciones diferentes de la siguiente manera.


aws timestream-query execute-scheduled-query --scheduled-query-arn arn_1 --invocation-time 1638316800 --profile profile --region us-east-1

aws timestream-query execute-scheduled-query --scheduled-query-arn arn_1 --invocation-time 1638324000 --profile profile --region us-east-1

aws timestream-query execute-scheduled-query --scheduled-query-arn arn_1 --invocation-time 1638331200 --profile profile --region us-east-1

aws timestream-query execute-scheduled-query --scheduled-query-arn arn_1 --invocation-time 1638338400 --profile profile --region us-east-1

aws timestream-query execute-scheduled-query --scheduled-query-arn arn_1 --invocation-time 1638345600 --profile profile --region us-east-1

aws timestream-query execute-scheduled-query --scheduled-query-arn arn_1 --invocation-time 1638352800 --profile profile --region us-east-1

Los seis comandos anteriores corresponden al cálculo programado que se ejecutó el 01-12-2021 00:00:00, 01-12-2021 02:00:00, 01-12-2021 04:0:00, 01-12-20211 06:00:00, 01-12-2021 08:00:00, y 01-12-2021 10:00:

Como alternativa, puede utilizar el cálculo MultiPT12HPerHrPerTimeseriesDPCountCatchUp, que se activó el 01-12-2021 13:00:00 en una ejecución para actualizar los agregados de todo el intervalo de tiempo de 12 horas. Por ejemplo, si arn_2 es el ARN de ese cálculo, puede ejecutar el siguiente comando desde la CLI.


aws timestream-query execute-scheduled-query --scheduled-query-arn arn_2 --invocation-time 1638363600 --profile profile --region us-east-1

Cabe señalar que, en el caso de un activador manual, puede utilizar una marca de tiempo para el parámetro de tiempo de invocación que no necesite estar alineada con las marcas de tiempo de ese activador automático. Por ejemplo, en el ejemplo anterior, activó el cálculo el 01-12-2021 13:00:00, aunque la programación automática solo se activa en las marcas de tiempo 01-12-2021 10:00:00, 01-12-2021 12:00:00, y 01-12-2021 00:00:00. Timestream para LiveAnalytics le ofrece la flexibilidad de activarlo con los valores adecuados según sea necesario para las operaciones manuales.

A continuación, presentamos algunas consideraciones importantes al utilizar la API ExecuteScheduledQuery.

Si va a activar varias de estas invocaciones, debe asegurarse de que estas no generen resultados en intervalos de tiempo superpuestos. Por ejemplo, en los ejemplos anteriores, hubo seis invocaciones. Cada invocación abarca un intervalo de tiempo de 2 horas y, por lo tanto, las marcas de tiempo de la invocación se distribuyeron en dos horas cada una para evitar cualquier superposición en las actualizaciones. Esto garantiza que los datos de la tabla derivada terminen en un estado que coincida con los agregados de la tabla de origen. Si no puede garantizar que los intervalos de tiempo no se superpongan, asegúrese de que las ejecuciones se activen de manera secuencial una tras otra. Si activa varias ejecuciones de forma simultánea y se superponen en los intervalos de tiempo, puede haber errores de activación en los informes de errores correspondientes a estas ejecuciones, por lo que es posible que aparezcan conflictos de versiones. A los resultados que generó una invocación de consulta programada se les asigna una versión en función del momento en que se activó la invocación. Por lo tanto, las filas que generaron las invocaciones más recientes tienen versiones superiores. Un registro de versión superior puede sobrescribir un registro de versión inferior. En el caso de las consultas programadas que se activan de manera automática, Timestream para LiveAnalytics administra automáticamente las programaciones para que no se produzcan estos problemas, incluso si las invocaciones posteriores tienen intervalos de tiempo superpuestos.
Como se indicó anteriormente, puede activar las invocaciones con cualquier valor de marca de tiempo para @scheduled_runtime. Por lo tanto, es su responsabilidad establecer los valores de manera adecuada para que los intervalos de tiempo adecuados se actualicen en la tabla derivada correspondiente a los intervalos en los que se actualizaron los datos en la tabla de origen.
También puede utilizar estos activadores manuales para las consultas programadas que estén en estado DESACTIVADO. Esto le permite definir las consultas especiales que no se ejecutan en un programa automático, ya que se encuentran en estado DESACTIVADO. Por el contrario, puede utilizar los activadores manuales para gestionar las correcciones de datos o los casos de uso de llegadas tardías.

Aviso JavaScript está desactivado o no está disponible en su navegador.

Para utilizar la documentación de AWS, debe estar habilitado JavaScript. Para obtener más información, consulte las páginas de ayuda de su navegador.

Convenciones del documento

Valores de la dimensión únicos

Completar los cálculos previos históricos