Contoh: Jendela Stagger

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

Pilih Data Streams (Aliran Data) di panel navigasi.

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.

Untuk menulis catatan ke Kinesis data stream di lingkungan produksi, sebaiknya gunakan Kinesis Producer 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 grup enam catatan dengan EVENT_TIME acak yang sama dan simbol ticker ke aliran, selama satu menit. Jaga skrip agar tetap berjalan agar Anda dapat menghasilkan skema aplikasi di langkah berikutnya.

 
import datetime
import json
import random
import time
import boto3

STREAM_NAME = "ExampleInputStream"


def get_data():
    event_time = datetime.datetime.utcnow() - datetime.timedelta(seconds=10)
    return {
        "EVENT_TIME": event_time.isoformat(),
        "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
    }


def generate(stream_name, kinesis_client):
    while True:
        data = get_data()
        # Send six records, ten seconds apart, with the same event time and ticker
        for _ in range(6):
            print(data)
            kinesis_client.put_record(
                StreamName=stream_name,
                Data=json.dumps(data),
                PartitionKey="partitionkey",
            )
            time.sleep(10)


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

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

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

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

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 dua 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).

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.

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_symbol    VARCHAR(4),
       ticker_count     INTEGER);
   
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
     INSERT INTO "DESTINATION_SQL_STREAM" 
       SELECT STREAM 
           EVENT_TIME, 
           TICKER,
           COUNT(TICKER) AS ticker_count
       FROM "SOURCE_SQL_STREAM_001"
       WINDOWED BY STAGGER (
               PARTITION BY TICKER, EVENT_TIME RANGE INTERVAL '1' MINUTE);

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.