Langkah 1 Buat Aliran Input dan Output - Panduan Developer Amazon Kinesis Data Analytics untuk Aplikasi SQL
Langkah 1.1: Buat Kinesis Data StreamsLangkah 1.2: Tulis Catatan Sampel ke Aliran Input

Untuk proyek baru, kami menyarankan Anda menggunakan Managed Service baru untuk Apache Flink Studio melalui Kinesis Data Analytics untuk Aplikasi SQL. Layanan Terkelola untuk Apache Flink Studio menggabungkan kemudahan penggunaan dengan kemampuan analitis tingkat lanjut, memungkinkan Anda membangun aplikasi pemrosesan aliran yang canggih dalam hitungan menit.

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

Langkah 1 Buat Aliran Input dan Output

Sebelum membuat aplikasi Amazon Kinesis Data Analytics untuk Latihan hotspots, Anda membuat dua Kinesis data streams. Konfigurasikan salah satu aliran sebagai sumber streaming untuk aplikasi Anda, dan aliran lainnya sebagai tujuan tempat Kinesis Data Analytics menyimpan output aplikasi Anda.

Langkah 1.1: Buat Kinesis Data Streams

Di bagian ini, Anda membuat dua Kinesis data streams: ExampleInputStream dan ExampleOutputStream.

Buat aliran data ini menggunakan konsol atau AWS CLI.

  • Buat aliran data ini menggunakan konsol:

    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), dan buat aliran dengan satu serpihan bernama ExampleInputStream.

    4. Ulangi langkah sebelumnya, yang membuat aliran dengan satu serpihan bernama ExampleOutputStream.

  • Buat aliran data ini menggunakan AWS CLI:

    • Buat aliran (ExampleInputStream dan ExampleOutputStream) menggunakan perintah create-stream AWS CLI Kinesis berikut. Untuk membuat aliran kedua, aplikasi mana yang akan digunakan untuk menulis output, jalankan perintah yang sama, yang mengubah nama aliran menjadi ExampleOutputStream.

      $ aws kinesis create-stream \
--stream-name ExampleInputStream \
--shard-count 1 \
--region us-west-2 \
--profile adminuser
                         
$ aws kinesis create-stream \
--stream-name ExampleOutputStream \
--shard-count 1 \
--region us-west-2 \
--profile adminuser

Langkah 1.2: Tulis Catatan Sampel ke Aliran Input

Dalam langkah ini, Anda menjalankan kode Python untuk terus membuat catatan sampel dan menulis ke aliran ExampleInputStream.

{"x": 7.921782426109737, "y": 8.746265312709893, "is_hot": "N"}
{"x": 0.722248626580026, "y": 4.648868803193405, "is_hot": "Y"}

  1. Instal Python dan pip.

    Untuk informasi tentang menginstal Python, lihat situs web Python.

    Anda dapat menginstal dependensi menggunakan pip. Untuk informasi tentang menginstal pip, lihat Penginstalan di situs web pip.

  2. Jalankan kode Phyton berikut. Kode ini melakukan hal berikut:

    • Membuat potensi hotspot di suatu tempat di bidang (X, Y).

    • Membuat satu kumpulan 1.000 poin untuk setiap hotspot. Dari titik-titik ini, 20 persen diklasterkan di sekitar hotspot. Sisanya dihasilkan secara acak di seluruh ruang.

    • Perintah put-record menulis catatan JSON ke aliran.

    penting

    Jangan unggah file ini ke server web karena berisi kredensial AWS Anda.

     
import json
from pprint import pprint
import random
import time
import boto3

STREAM_NAME = "ExampleInputStream"


def get_hotspot(field, spot_size):
    hotspot = {
        "left": field["left"] + random.random() * (field["width"] - spot_size),
        "width": spot_size,
        "top": field["top"] + random.random() * (field["height"] - spot_size),
        "height": spot_size,
    }
    return hotspot


def get_record(field, hotspot, hotspot_weight):
    rectangle = hotspot if random.random() < hotspot_weight else field
    point = {
        "x": rectangle["left"] + random.random() * rectangle["width"],
        "y": rectangle["top"] + random.random() * rectangle["height"],
        "is_hot": "Y" if rectangle is hotspot else "N",
    }
    return {"Data": json.dumps(point), "PartitionKey": "partition_key"}


def generate(
    stream_name, field, hotspot_size, hotspot_weight, batch_size, kinesis_client
):
    """
    Generates points used as input to a hotspot detection algorithm.
    With probability hotspot_weight (20%), a point is drawn from the hotspot;
    otherwise, it is drawn from the base field. The location of the hotspot
    changes for every 1000 points generated.
    """
    points_generated = 0
    hotspot = None
    while True:
        if points_generated % 1000 == 0:
            hotspot = get_hotspot(field, hotspot_size)
        records = [
            get_record(field, hotspot, hotspot_weight) for _ in range(batch_size)
        ]
        points_generated += len(records)
        pprint(records)
        kinesis_client.put_records(StreamName=stream_name, Records=records)

        time.sleep(0.1)


if __name__ == "__main__":
    generate(
        stream_name=STREAM_NAME,
        field={"left": 0, "width": 10, "top": 0, "height": 10},
        hotspot_size=1,
        hotspot_weight=0.2,
        batch_size=10,
        kinesis_client=boto3.client("kinesis"),
    )

Langkah Selanjutnya

Langkah 2: Buat Aplikasi Kinesis Data Analytics