Menangani data yang datang terlambat - Amazon Timestream
Pertanyaan mengejar ketinggalan terjadwalEksekusi manual untuk data kedatangan terlambat yang tidak dapat diprediksi

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Menangani data yang datang terlambat

Anda mungkin memiliki skenario di mana Anda dapat memiliki data yang datang terlambat secara signifikan, misalnya, waktu ketika data dicerna ke Timestream untuk secara signifikan LiveAnalytics tertunda dibandingkan dengan stempel waktu yang terkait dengan baris yang dicerna. Dalam contoh sebelumnya, Anda telah melihat bagaimana Anda dapat menggunakan rentang waktu yang ditentukan oleh parameter @scheduled_runtime untuk memperhitungkan beberapa data yang terlambat tiba. Namun, jika Anda memiliki kasus penggunaan di mana data dapat ditunda oleh jam atau hari, Anda mungkin memerlukan pola yang berbeda untuk memastikan pra-perhitungan Anda dalam tabel turunan diperbarui dengan tepat untuk mencerminkan data yang datang terlambat tersebut. Untuk informasi umum tentang data yang datang terlambat, lihat. Menulis data (sisipan dan upserts)

Berikut ini Anda akan melihat dua cara berbeda untuk mengatasi data yang datang terlambat ini.

  • Jika Anda memiliki penundaan yang dapat diprediksi dalam kedatangan data Anda, maka Anda dapat menggunakan perhitungan terjadwal “catch-up” lain untuk memperbarui agregat Anda untuk data yang terlambat tiba.

  • Jika Anda memiliki penundaan yang tidak dapat diprediksi atau data kedatangan terlambat sesekali, Anda dapat menggunakan eksekusi manual untuk memperbarui tabel turunan.

Diskusi ini mencakup skenario untuk kedatangan data yang terlambat. Namun, prinsip yang sama berlaku untuk koreksi data, di mana Anda telah memodifikasi data dalam tabel sumber Anda dan Anda ingin memperbarui agregat dalam tabel turunan Anda.

Pertanyaan mengejar ketinggalan terjadwal

Kueri mengumpulkan data yang tiba tepat waktu

Di bawah ini adalah pola yang akan Anda lihat bagaimana Anda dapat menggunakan cara otomatis untuk memperbarui agregat Anda jika Anda memiliki penundaan yang dapat diprediksi dalam kedatangan data Anda. Pertimbangkan salah satu contoh sebelumnya dari perhitungan terjadwal pada data real-time di bawah ini. Perhitungan terjadwal ini menyegarkan tabel turunan setiap 30 menit sekali dan sudah memperhitungkan data hingga satu jam tertunda.

{
    "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": "******"
}

Kueri catch-up memperbarui agregat untuk data yang datang terlambat

Sekarang jika Anda mempertimbangkan kasus bahwa data Anda dapat ditunda sekitar 12 jam. Di bawah ini adalah varian dari kueri yang sama. Namun, perbedaannya adalah bahwa ia menghitung agregat pada data yang tertunda hingga 12 jam dibandingkan dengan saat perhitungan terjadwal sedang dipicu. Misalnya, Anda melihat kueri pada contoh di bawah ini, rentang waktu yang ditargetkan kueri ini adalah antara 2 jam hingga 14 jam sebelum ketika kueri dipicu. Selain itu, jika Anda melihat ekspresi jadwal cron (0 0,12 * *? *), itu akan memicu perhitungan pada 00:00 UTC dan 12:00 UTC setiap hari. Oleh karena itu, ketika kueri dipicu pada 2021-12-01 00:00:00, maka kueri memperbarui agregat dalam rentang waktu 2021-11-30 10:00:00 hingga 2021-11-30 22:00:00. Kueri terjadwal menggunakan semantik upsert yang mirip dengan Timestream untuk LiveAnalytics penulisan di mana kueri penangkapan ini akan memperbarui nilai agregat dengan nilai yang lebih baru jika ada data yang datang terlambat di jendela atau jika agregat yang lebih baru ditemukan (misalnya, pengelompokan baru muncul di agregat ini yang tidak ada saat perhitungan terjadwal asli dipicu), maka agregat baru akan dimasukkan ke dalam tabel turunan. Demikian pula, ketika instance berikutnya dipicu pada 2021-12-01 12:00:00, maka instance itu akan memperbarui agregat dalam kisaran 2021-11-30 22:00:00 hingga 2021-12-01 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": "******"
}

Contoh sebelumnya ini adalah ilustrasi dengan asumsi kedatangan terlambat Anda dibatasi hingga 12 jam dan tidak apa-apa untuk memperbarui tabel turunan setiap 12 jam sekali untuk data yang tiba lebih lambat dari jendela waktu nyata. Anda dapat menyesuaikan pola ini untuk memperbarui tabel turunan Anda sekali setiap jam sehingga tabel turunan Anda mencerminkan data yang datang terlambat lebih cepat. Demikian pula, Anda dapat menyesuaikan rentang waktu menjadi lebih tua dari 12 jam, misalnya, sehari atau bahkan seminggu atau lebih, untuk menangani data kedatangan terlambat yang dapat diprediksi.

Eksekusi manual untuk data kedatangan terlambat yang tidak dapat diprediksi

Mungkin ada contoh di mana Anda memiliki data kedatangan terlambat yang tidak dapat diprediksi atau Anda membuat perubahan pada data sumber dan memperbarui beberapa nilai setelah fakta. Dalam semua kasus seperti itu, Anda dapat memicu kueri terjadwal secara manual untuk memperbarui tabel turunan. Di bawah ini adalah contoh bagaimana Anda dapat mencapai ini.

Asumsikan bahwa Anda memiliki kasus penggunaan di mana Anda memiliki perhitungan yang ditulis ke tabel turunan dp_per_timeseries_per_hr. Data dasar Anda dalam tabel devops diperbarui pada rentang waktu 2021-11-30 23:00:00 - 2021-12-01 00:00:00. Ada dua kueri terjadwal berbeda yang dapat digunakan untuk memperbarui tabel turunan ini: Multi PT3 0 mPerHr PerTimeseries DPCount dan Multi PT12 HPer HrPerTimeseries DPCountCatchUp. Setiap perhitungan terjadwal yang Anda buat di Timestream LiveAnalytics memiliki ARN unik yang Anda peroleh saat membuat perhitungan atau saat Anda melakukan operasi daftar. Anda dapat menggunakan ARN untuk perhitungan dan nilai untuk parameter @scheduled_runtime yang diambil oleh kueri untuk melakukan operasi ini.

Asumsikan bahwa perhitungan untuk Multi PT3 0 mPerHr PerTimeseries DPCount memiliki ARN arn_1 dan Anda ingin menggunakan perhitungan ini untuk memperbarui tabel turunan. Karena komputasi terjadwal sebelumnya memperbarui agregat 1 jam sebelum dan 1 jam setelah nilai @scheduled_runtime, Anda dapat mencakup rentang waktu untuk pembaruan (2021-11-30 23:00:00 - 2021-12-01 00:00:00) menggunakan nilai 2021-12-01 00:00:00 untuk parameter @scheduled_runtime. Anda dapat menggunakan ExecuteScheduledQuery API untuk meneruskan ARN perhitungan ini dan nilai parameter waktu dalam epoch seconds (dalam UTC) untuk mencapai ini. Di bawah ini adalah contoh menggunakan AWS CLI dan Anda dapat mengikuti pola yang sama menggunakan salah satu yang SDKs didukung oleh Timestream untuk. LiveAnalytics

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

Pada contoh sebelumnya, profil adalah AWS profil yang memiliki hak istimewa yang sesuai untuk melakukan panggilan API ini dan 1638316800 sesuai dengan epoch second untuk 2021-12-01 00:00:00. Pemicu manual ini berperilaku hampir seperti pemicu otomatis dengan asumsi sistem memicu pemanggilan ini pada periode waktu yang diinginkan.

Jika Anda memiliki pembaruan dalam periode waktu yang lebih lama, katakanlah data dasar telah diperbarui untuk 2021-11-30 23:00:00 - 2021-12-01 11:00:00, maka Anda dapat memicu kueri sebelumnya beberapa kali untuk mencakup seluruh rentang waktu ini. Misalnya, Anda dapat melakukan enam eksekusi yang berbeda sebagai berikut.

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

Enam perintah sebelumnya sesuai dengan perhitungan terjadwal yang dipanggil pada 2021-12-01 00:00:00, 2021-12-01 02:00:00, 2021-12-01 04:0:00, 2021-12-01 06:00:00, 2021-12-01 08:00:00, dan 2021-12-01 10:00:

Atau, Anda dapat menggunakan komputasi Multi PT12 HPer HrPerTimeseries DPCount CatchUp dipicu pada 2021-12-01 13:00:00 untuk satu eksekusi guna memperbarui agregat untuk seluruh rentang waktu 12 jam. Misalnya, jika arn_2 adalah ARN untuk perhitungan itu, Anda dapat menjalankan perintah berikut dari CLI.

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

Perlu dicatat bahwa untuk pemicu manual, Anda dapat menggunakan stempel waktu untuk parameter waktu pemanggilan yang tidak perlu disejajarkan dengan stempel waktu pemicu otomatis tersebut. Misalnya, pada contoh sebelumnya, Anda memicu perhitungan pada waktu 2021-12-01 13:00:00 meskipun jadwal otomatis hanya terpicu pada stempel waktu 2021-12-01 10:00:00, 2021-12-01 12:00:00, dan 2021-12-02 00:00:00. Timestream untuk LiveAnalytics memberi Anda fleksibilitas untuk memicunya dengan nilai yang sesuai sesuai kebutuhan untuk operasi manual Anda.

Berikut ini adalah beberapa pertimbangan penting saat menggunakan ExecuteScheduledQuery API.

  • Jika Anda memicu beberapa pemanggilan ini, Anda perlu memastikan bahwa pemanggilan ini tidak menghasilkan hasil dalam rentang waktu yang tumpang tindih. Misalnya, dalam contoh sebelumnya, ada enam doa. Setiap pemanggilan mencakup rentang waktu 2 jam, dan karenanya stempel waktu pemanggilan disebarkan masing-masing dua jam untuk menghindari tumpang tindih dalam pembaruan. Ini memastikan bahwa data dalam tabel turunan berakhir dalam keadaan yang cocok adalah agregat dari tabel sumber. Jika Anda tidak dapat memastikan rentang waktu yang tidak tumpang tindih, maka pastikan eksekusi ini dipicu secara berurutan satu demi satu. Jika Anda memicu beberapa eksekusi secara bersamaan yang tumpang tindih dalam rentang waktunya, maka Anda dapat melihat kegagalan pemicu di mana Anda mungkin melihat konflik versi dalam laporan kesalahan untuk eksekusi ini. Hasil yang dihasilkan oleh pemanggilan kueri terjadwal diberi versi berdasarkan kapan pemanggilan dipicu. Oleh karena itu, baris yang dihasilkan oleh pemanggilan yang lebih baru memiliki versi yang lebih tinggi. Catatan versi yang lebih tinggi dapat menimpa catatan versi yang lebih rendah. Untuk kueri terjadwal yang dipicu secara otomatis, Timestream untuk LiveAnalytics secara otomatis mengelola jadwal sehingga Anda tidak melihat masalah ini bahkan jika pemanggilan berikutnya memiliki rentang waktu yang tumpang tindih.

  • disebutkan sebelumnya, Anda dapat memicu pemanggilan dengan nilai stempel waktu apa pun untuk @scheduled_runtime. Jadi, Anda bertanggung jawab untuk mengatur nilai dengan tepat sehingga rentang waktu yang sesuai diperbarui dalam tabel turunan yang sesuai dengan rentang di mana data diperbarui di tabel sumber.

  • Anda juga dapat menggunakan pemicu manual ini untuk kueri terjadwal yang berada dalam status DISABLED. Ini memungkinkan Anda untuk menentukan kueri khusus yang tidak dieksekusi dalam jadwal otomatis, karena mereka berada dalam status DISABLED. Sebaliknya, Anda dapat menggunakan pemicu manual pada mereka untuk mengelola koreksi data atau kasus penggunaan keterlambatan kedatangan.