Contoh: Jendela Tumbling Menggunakan Stempel Waktu Peristiwa - Panduan Pengembang Amazon Kinesis Data Analytics SQL untuk Aplikasi
Langkah 1: Buat Kinesis Data StreamLangkah 2: Buat Aplikasi Kinesis Data Analytics

Setelah mempertimbangkan dengan cermat, kami memutuskan untuk menghentikan Amazon Kinesis Data Analytics SQL untuk aplikasi dalam dua langkah:

1. Mulai 15 Oktober 2025, Anda tidak akan dapat membuat Kinesis Data Analytics SQL baru untuk aplikasi.

2. Kami akan menghapus aplikasi Anda mulai 27 Januari 2026. Anda tidak akan dapat memulai atau mengoperasikan Amazon Kinesis Data Analytics Anda SQL untuk aplikasi. Support tidak akan lagi tersedia untuk Amazon Kinesis Data Analytics SQL sejak saat itu. Untuk informasi selengkapnya, lihat Amazon Kinesis Data Analytics SQL untuk penghentian Aplikasi.

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

Contoh: Jendela Tumbling Menggunakan Stempel Waktu Peristiwa

Ketika kueri jendela memproses setiap jendela dengan cara yang tidak tumpang tindih, jendela disebut sebagai jendela tumbling. Untuk detailnya, lihat Jendela Tumbling (Agregasi Menggunakan GROUP BY). Contoh Amazon Kinesis Data Analytics menunjukkan jendela tumbling yang menggunakan stempel waktu peristiwa, yang merupakan stempel waktu yang dibuat pengguna yang disertakan dalam data streaming. Contoh tersebut menggunakan pendekatan ini daripada hanya menggunakan ROWTIME, yang merupakan stempel waktu yang dibuat Kinesis Data Analytics saat aplikasi menerima catatan. Anda akan menggunakan stempel waktu peristiwa dalam data streaming jika Anda ingin membuat agregasi berdasarkan kapan peristiwa terjadi, bukan ketika peristiwa diterima oleh aplikasi. Dalam contoh ini, nilai ROWTIME memicu agregasi setiap menit, dan catatan dikumpulkan oleh kedua ROWTIME dan waktu peristiwa yang disertakan.

Dalam contoh ini, Anda menulis catatan berikut ke Amazon Kinesis stream. Nilai EVENT_TIME sebelumnya diatur ke 5 detik, untuk menyimulasikan keterlambatan pemrosesan dan transmisi yang mungkin membuat penundaan sejak peristiwa terjadi, hingga saat catatan tersebut diserap ke Kinesis Data Analytics.

{"EVENT_TIME": "2018-06-13T14:11:05.766191", "TICKER": "TBV", "PRICE": 43.65}
{"EVENT_TIME": "2018-06-13T14:11:05.848967", "TICKER": "AMZN", "PRICE": 35.61}
{"EVENT_TIME": "2018-06-13T14:11:05.931871", "TICKER": "MSFT", "PRICE": 73.48}
{"EVENT_TIME": "2018-06-13T14:11:06.014845", "TICKER": "AMZN", "PRICE": 18.64}
...

Anda kemudian membuat aplikasi Kinesis Data Analytics di AWS Management Console, dengan Kinesis data stream sebagai sumber streaming. Proses penemuan membaca catatan sampel pada sumber streaming dan menyimpulkan skema dalam aplikasi dengan tiga kolom (EVENT_TIME, TICKER, dan PRICE) seperti yang ditunjukkan berikut ini.

Tangkapan layar konsol yang menampilkan skema dalam aplikasi dengan waktu peristiwa, ticker, dan kolom harga.

Anda menggunakan kode aplikasi dengan fungsi MIN dan MAX untuk membuat jendela agregasi data. Anda selanjutnya memasukkan data yang dihasilkan ke aliran dalam aplikasi lainnya, seperti yang ditunjukkan dalam tangkapan layar bawah ini:

Tangkapan layar konsol yang menampilkan data yang dihasilkan di aliran dalam aplikasi.

Dalam prosedur berikut, Anda membuat aplikasi Kinesis Data Analytics yang menggabungkan nilai dalam aliran input di jendela tumbling berdasarkan waktu peristiwa.

Langkah 1: Buat Kinesis Data Stream

Buat Amazon Kinesis data stream dan isi catatan sebagai berikut:

  1. Masuk ke AWS Management Console dan buka konsol Kinesis di https://console.aws.amazon.com/kinesis.

  2. Pilih Data Streams (Aliran Data) di panel navigasi.

  3. Pilih Create Kinesis stream (Buat Aliran Kinesis), lalu buat aliran dengan satu serpihan. Untuk informasi selengkapnya, lihat Buat Aliran di Panduan Developer Amazon Kinesis Data Streams.

  4. Untuk menulis catatan ke Kinesis data stream di lingkungan produksi, sebaiknya gunakan Kinesis Client Library atau API Kinesis Data Streams. Untuk kemudahan, contoh ini menggunakan skrip Python berikut untuk menghasilkan catatan. Jalankan kode untuk mengisi catatan ticker sampel. Kode sederhana ini terus menulis catatan ticker acak ke aliran. Jaga skrip agar tetap berjalan agar Anda dapat menghasilkan skema aplikasi di langkah berikutnya.

     
import datetime
import json
import random
import boto3

STREAM_NAME = "ExampleInputStream"


def get_data():
    return {
        "EVENT_TIME": datetime.datetime.now().isoformat(),
        "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
        "PRICE": round(random.random() * 100, 2),
    }


def generate(stream_name, kinesis_client):
    while True:
        data = get_data()
        print(data)
        kinesis_client.put_record(
            StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
        )


if __name__ == "__main__":
    generate(STREAM_NAME, boto3.client("kinesis"))

Langkah 2: Buat Aplikasi Kinesis Data Analytics

Buat aplikasi Amazon Kinesis Data Analytics seperti berikut:

  1. Buka Layanan Terkelola untuk konsol Apache Flink di https://console.aws.amazon.com/kinesisanalytics.

  2. Pilih Create application (Buat aplikasi), masukkan nama aplikasi, dan pilih Create application (Buat aplikasi).

  3. Pada halaman detail aplikasi, pilih Connect data streaming (Sambungkan data streaming) untuk menyambungkan ke sumber.

  4. Di halaman Sambungkan ke sumber, lakukan hal berikut:

    1. Pilih aliran yang Anda buat di bagian sebelumnya.

    2. Pilih Discover Schema (Temukan Skema). Tunggu hingga konsol menampilkan skema yang disimpulkan dan catatan sampel yang digunakan untuk menyimpulkan skema untuk aliran dalam aplikasi yang dibuat. Skema yang disimpulkan memiliki tiga kolom.

    3. Pilih Edit Schema (Edit Skema). Ubah Column type (Tipe kolom) dari kolom EVENT_TIME ke TIMESTAMP.

    4. Pilih Save schema and update stream samples (Simpan skema dan perbarui sampel aliran). Setelah konsol menyimpan skema, pilih Exit (Keluar).

    5. Jangan pilih Save and continue (Simpan dan lanjutkan).

  5. Di halaman detail aplikasi, pilih Go to SQL editor (Buka editor SQL). Untuk memulai aplikasi, pilih Yes, start application (Ya, mulai aplikasi) di kotak dialog yang muncul.

  6. Di editor SQL, tulis kode aplikasi, dan verifikasi hasilnya sebagai berikut:

    1. Salin kode aplikasi berikut dan tempelkan ke editor.

      CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (EVENT_TIME timestamp, TICKER VARCHAR(4), min_price REAL, max_price REAL);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
  INSERT INTO "DESTINATION_SQL_STREAM" 
    SELECT STREAM STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND),
        TICKER,
         MIN(PRICE) AS MIN_PRICE,
         MAX(PRICE) AS MAX_PRICE
    FROM    "SOURCE_SQL_STREAM_001"
    GROUP BY TICKER, 
             STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND), 
             STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND);

    2. Pilih Save and run SQL (Simpan dan jalankan SQL).

      Di tab Real-time analytics (Analitik waktu nyata), Anda dapat melihat semua aliran dalam aplikasi yang dibuat aplikasi dan memverifikasi data.