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