Buat notebook Studio dengan Amazon MSK

1. Buka VPC konsol Amazon di https://console.aws.amazon.com/vpc/.

2. Pilih NATGateway dari bilah navigasi kiri.

3. Pada halaman NATGateway, pilih NATCreate Gateway.

4. Pada halaman Create NAT Gateway, berikan nilai berikut:

   Nama - opsional	ZeppelinGateway
   Subnet	AWS KafkaTutorialSubnet1
   ID alokasi IP elastis	Pilih IP Elastis yang tersedia. Jika tidak ada Elastic IPs yang tersedia, pilih Alokasikan IP Elastis, lalu pilih IP Elasic yang dibuat konsol.

   Pilih Buat NAT Gateway.

5. Di bilah navigasi sebelah kiri, pilih Route Tables (Tabel Rute).

6. Pilih Create Route Table (Buat Tabel Rute).

7. Di halaman Create route table (Buat tabel rute), berikan informasi berikut:

   • Name tag (Tanda nama): ZeppelinRouteTable

   • VPC: Pilih Anda VPC (mis. AWS KafkaTutorialVPC).

   Pilih Buat.

8. Dalam daftar tabel rute, pilih ZeppelinRouteTable. Pilih tab Routes (Rute), dan pilih Edit routes (Edit rute).

9. Di halaman Edit Rute, pilih Add route (Tambahkan rute).

10. Di Untuk Tujuan, masukkan 0.0.0.0/0. Untuk Target, pilih NATGateway, ZeppelinGateway. Pilih Save Routes (Simpan Rute). Pilih Close (Tutup).

11. Pada halaman Tabel Rute, dengan ZeppelinRouteTabledipilih, pilih tab Asosiasi Subnet. Pilih Edit subnet associations (Edit asosiasi subnet).

12. Di halaman Edit asosiasi subnet, pilih AWS KafkaTutorialSubnet2 dan AWS KafkaTutorialSubnet3. Pilih Simpan.

Buat Koneksi

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

2. Jika Anda belum memiliki AWS Glue database, pilih Database dari bilah navigasi kiri. Pilih Add database (Tambahkan basis data). Di jendela Add database (Tambahkan basis data), masukkan default untuk Database name (Nama basis data). Pilih Create (Buat).

3. Pilih Connections (Koneksi) dari bilah navigasi sebelah kiri. Pilih Add Connection (Tambahkan Koneksi).

4. Di jendela Tambahkan Koneksi, berikan nilai berikut:

   • Untuk Connection name (Nama koneksi), masukkan ZeppelinConnection.

   • Untuk Connection type (Tipe koneksi), pilih Kafka.

   • Untuk server bootstrap Kafka URLs, berikan string broker bootstrap untuk cluster Anda. Anda bisa mendapatkan broker bootstrap dari MSK konsol, atau dengan memasukkan CLI perintah berikut:

     aws kafka get-bootstrap-brokers --region us-east-1 --cluster-arn ClusterArn

   • Hapus centang pada kotak centang Memerlukan SSL koneksi.

   Pilih Berikutnya.

5. Di VPChalaman, berikan nilai-nilai berikut:

   • Untuk VPC, pilih nama Anda VPC (mis AWS KafkaTutorialVPC.)

   • Untuk Subnet, pilih AWS KafkaTutorialSubnet2.

   • Untuk Security groups (Grup keamanan), pilih semua grup yang tersedia.

   Pilih Berikutnya.

6. Di halaman Properti koneksi / Akses koneksi, pilih Finish (Selesai).

Buat Tabel

1. Di bilah navigasi sebelah kiri, pilih Tables (Tabel). Di halaman Tabel, pilih Add tables (Tambahkan tabel), Add table manually (Tambahkan tabel secara manual).

2. Di halaman Set up your table's properties (Siapkan properti tabel Anda), masukkan stock untuk Table name (Nama tabel). Pastikan Anda memilih basis data yang Anda buat sebelumnya. Pilih Berikutnya.

3. Di halaman Tambahkan penyimpanan data, pilih Kafka. Untuk nama Topik, masukkan nama topik Anda (mis. AWS KafkaTutorialTopic). Untuk Koneksi, pilih ZeppelinConnection.

4. Di halaman Klasifikasi, pilih JSON. Pilih Berikutnya.

5. Di halaman Tentukan skema, pilih Add Column (Tambahkan kolom) untuk menambahkan kolom. Tambahkan kolom dengan properti berikut:

   Nama kolom	Tipe data
   ticker	string
   price	double

   Pilih Berikutnya.

6. Di halaman berikutnya, verifikasi pengaturan Anda, dan pilih Finish (Selesai).

7. Pilih tabel yang baru dibuat dari daftar tabel.

8. Pilih Edit tabel dan tambahkan properti berikut:

   • kunci:managed-flink.proctime, nilai: proctime

   • kunci:flink.properties.group.id, nilai: test-consumer-group

   • kunci:flink.properties.auto.offset.reset, nilai: latest

   • kunci:classification, nilai: json

   Tanpa pasangan kunci/nilai ini, notebook Flink mengalami kesalahan.

9. Pilih Terapkan.

1. Buka Layanan Terkelola untuk konsol Apache Flink di https://console.aws.amazon.com/managed-flink/ rumah? region=us-east-1#/aplikasi/dasbor.

2. Di halaman Managed Service for Apache Flink Apache Applications, pilih tab Studio. Pilih Create Studio notebook (Buat notebook Studio).

   catatan

   Untuk membuat notebook Studio dari konsol Amazon MSK atau Kinesis Data Streams, pilih cluster MSK Amazon input atau aliran data Kinesis, lalu pilih Memproses data secara real time.

3. Di halaman Buat notebook Studio, berikan informasi berikut:

   • Masukkan MyNotebook untuk Studio notebook Name (Nama notebook Studio).

   • Pilih default untuk Basis data AWS Glue.

   Pilih Create Studio notebook (Buat notebook Studio).

4. Di MyNotebookhalaman, pilih tab Konfigurasi. Di bagian Jaringan, pilih Edit.

5. Di MyNotebook halaman Edit jaringan untuk, pilih VPCkonfigurasi berdasarkan MSK klaster Amazon. Pilih MSK klaster Amazon Anda untuk Amazon MSK Cluster. Pilih Simpan perubahan.

6. Di MyNotebookhalaman, pilih Jalankan. Tunggu Status hingga menampilkan Running (Berjalan).

1. Pastikan bahwa Anda memiliki informasi berikut. Anda perlu nilai-nilai ini untuk membuat aplikasi Anda.

   • ID akun Anda.

   • ID subnet IDs dan grup keamanan untuk Amazon VPC yang berisi MSK cluster Amazon Anda.

2. Buat file bernama create.json dengan konten berikut. Ganti nilai placeholder dengan informasi Anda.

   {
       "ApplicationName": "MyNotebook",
       "RuntimeEnvironment": "ZEPPELIN-FLINK-3_0",
       "ApplicationMode": "INTERACTIVE",
       "ServiceExecutionRole": "arn:aws:iam::AccountID:role/ZeppelinRole",
       "ApplicationConfiguration": {
           "ApplicationSnapshotConfiguration": {
               "SnapshotsEnabled": false
           },
           "VpcConfigurations": [
               {
                   "SubnetIds": [
                       "SubnetID 1",
                       "SubnetID 2",
                       "SubnetID 3"
                   ],
                   "SecurityGroupIds": [
                       "VPC Security Group ID"
                   ]
               }
           ],
           "ZeppelinApplicationConfiguration": {
               "CatalogConfiguration": {
                   "GlueDataCatalogConfiguration": {
                       "DatabaseARN": "arn:aws:glue:us-east-1:AccountID:database/default"
                   }
               }
           }
       }
   }
   

3. Jalankan perintah berikut untuk membuat aplikasi Anda.

   aws kinesisanalyticsv2 create-application --cli-input-json file://create.json 
   

4. Setelah perintah selesai, Anda akan melihat output yang serupa dengan yang berikut, yang menampilkan detail untuk notebook Studio baru Anda:

   {
       "ApplicationDetail": {
           "ApplicationARN": "arn:aws:kinesisanalyticsus-east-1:012345678901:application/MyNotebook",
           "ApplicationName": "MyNotebook",
           "RuntimeEnvironment": "ZEPPELIN-FLINK-3_0",
           "ApplicationMode": "INTERACTIVE",
           "ServiceExecutionRole": "arn:aws:iam::012345678901:role/ZeppelinRole",
   ...
   

5. Jalankan perintah berikut untuk memulai aplikasi Anda. Ganti nilai sampel dengan ID akun Anda.

   aws kinesisanalyticsv2 start-application --application-arn arn:aws:kinesisanalyticsus-east-1:012345678901:application/MyNotebook\
   

1. Connect ke EC2 klien Amazon Anda.

2. Jalankan perintah berikut untuk menginstal Python versi 3, Pip, dan Kafka untuk paket Python, dan mengonfirmasi tindakan:

   sudo yum install python37
   curl -O https://bootstrap.pypa.io/get-pip.py
   python3 get-pip.py --user
   pip install kafka-python

3. Konfigurasikan AWS CLI pada mesin klien Anda dengan memasukkan perintah berikut:

   aws configure

   Berikan kredensial akun Anda, dan us-east-1 untuk region.

4. Buat file bernama stock.py dengan konten berikut. Ganti nilai sampel dengan string Bootstrap Brokers MSK klaster Amazon Anda, dan perbarui nama topik jika topik Anda bukan AWS KafkaTutorialTopic:

   from kafka import KafkaProducer
   import json
   import random
   from datetime import datetime
   
   BROKERS = "<<Bootstrap Broker List>>"
   producer = KafkaProducer(
       bootstrap_servers=BROKERS,
       value_serializer=lambda v: json.dumps(v).encode('utf-8'),
       retry_backoff_ms=500,
       request_timeout_ms=20000,
       security_protocol='PLAINTEXT')
   
   
   def getStock():
       data = {}
       now = datetime.now()
       str_now = now.strftime("%Y-%m-%d %H:%M:%S")
       data['event_time'] = str_now
       data['ticker'] = random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV'])
       price = random.random() * 100
       data['price'] = round(price, 2)
       return data
   
   
   while True:
       data =getStock()
       # print(data)
       try:
           future = producer.send("AWSKafkaTutorialTopic", value=data)
           producer.flush()
           record_metadata = future.get(timeout=10)
           print("sent event to Kafka! topic {} partition {} offset {}".format(record_metadata.topic, record_metadata.partition, record_metadata.offset))
       except Exception as e:
           print(e.with_traceback())

5. Jalankan skrip dengan perintah berikut:

   $ python3 stock.py

6. Biarkan skrip berjalan saat Anda menyelesaikan bagian berikut.

1. Buka Layanan Terkelola untuk konsol Apache Flink di https://console.aws.amazon.com/managed-flink/ rumah? region=us-east-1#/aplikasi/dasbor.

2. Pada halaman Managed Service for Apache Flink Apache Applications, pilih tab notebook Studio. Pilih MyNotebook.

3. Di MyNotebookhalaman, pilih Buka di Apache Zeppelin.

   Antarmuka Apache Zeppelin terbuka di tab baru.

4. Di halaman Selamat Datang di Zeppelin!, pilih Zeppelin new note (Catatan baru Zeppelin).

5. Di halaman Zeppelin Note (Catatan Zeppelin), masukkan kueri berikut ke dalam catatan baru:

   %flink.ssql(type=update)
   select * from stock

   Pilih ikon jalankan.

   Aplikasi ini menampilkan data dari MSK cluster Amazon.