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:
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), dan buat aliran dengan satu serpihan bernama
ExampleInputStream
. -
Ulangi langkah sebelumnya, yang membuat aliran dengan satu serpihan bernama
ExampleOutputStream
.
-
Buat aliran data ini menggunakan AWS CLI:
-
Buat aliran (
ExampleInputStream
danExampleOutputStream
) menggunakan perintahcreate-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 menjadiExampleOutputStream
.$ aws kinesis create-stream \ --stream-name
ExampleInputStream
\ --shard-count 1 \ --region us-west-2 \ --profile adminuser $ aws kinesis create-stream \ --stream-nameExampleOutputStream
\ --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"}
-
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. -
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