Membuat dan menjalankan Managed Service untuk Apache Flink untuk aplikasi Python

Untuk membuat aliran data AWS CLI

1. Untuk membuat stream (ExampleInputStream) pertama, gunakan perintah Amazon Kinesis create-stream AWS CLI berikut.

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-east-1 

2. Untuk membuat aliran kedua yang digunakan aplikasi untuk menulis output, jalankan perintah yang sama, yang mengubah nama aliran menjadi ExampleOutputStream.

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-east-1 

1. Kloning repositori jarak jauh menggunakan perintah berikut:

   git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git

2. Buka direktori ./python/GettingStarted tersebut.

Untuk menjalankan aplikasi Anda dari baris perintah

1. Pastikan bahwa lingkungan Python mandiri seperti Conda VirtualEnv atau tempat Anda menginstal pustaka Python Flink saat ini aktif.

2. Pastikan Anda berlari mvn package setidaknya satu kali.

3. Atur variabel lingkungan IS_LOCAL = true:

   $ export IS_LOCAL=true

4. Jalankan aplikasi sebagai skrip Python biasa.

   $python main.py

Untuk menjalankan aplikasi dari dalam IDE

1. IDEKonfigurasikan Anda untuk menjalankan main.py skrip dengan konfigurasi berikut:

   1. Gunakan lingkungan Python mandiri seperti Conda VirtualEnv atau tempat Anda menginstal perpustakaan. PyFlink

   2. Gunakan AWS kredensil untuk mengakses input dan output Kinesis aliran data.

   3. Atur IS_LOCAL = true.

2. Proses yang tepat untuk mengatur konfigurasi run tergantung pada Anda IDE dan bervariasi.

3. Ketika Anda telah mengatur AndaIDE, jalankan skrip Python dan gunakan tooling yang disediakan oleh Anda IDE saat aplikasi sedang berjalan.

Untuk mengunggah JAR file kode aplikasi

1. Buka konsol Amazon S3 di. https://console.aws.amazon.com/s3/

2. Pilih bucket yang sebelumnya Anda buat untuk kode aplikasi.

3. Pilih Unggah.

4. Pilih Tambahkan file.

5. Arahkan ke file.zip yang dihasilkan pada langkah sebelumnya:target/managed-flink-pyflink-getting-started-1.0.0.zip.

6. Pilih Unggah tanpa mengubah pengaturan lainnya.

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

2. Verifikasi bahwa Wilayah yang benar dipilih: US East (Virginia N.) us-east-1.

3. Buka menu sisi kanan dan pilih aplikasi Apache Flink dan kemudian Buat aplikasi streaming. Atau, pilih Buat aplikasi streaming dari bagian Memulai di halaman awal.

4. Pada halaman Buat aplikasi streaming:

   • Untuk Memilih metode untuk mengatur aplikasi pemrosesan aliran, pilih Buat dari awal.

   • Untuk konfigurasi Apache Flink, versi Application Flink, pilih Apache Flink 1.19.

   • Untuk konfigurasi Aplikasi:

     • Untuk Application name (Nama aplikasi), masukkan MyApplication.

     • Untuk Description (Deskripsi), masukkan My Python test app.

     • Di Akses ke sumber daya aplikasi, pilih Buat/IAMperbarui peran kinesis-analytics- MyApplication -us-east-1 dengan kebijakan yang diperlukan.

   • Untuk Template untuk pengaturan aplikasi:

     • Untuk Template, pilih Development.

   • Pilih Buat aplikasi streaming.

Untuk mengedit IAM kebijakan untuk menambahkan izin bucket S3

1. Buka IAM konsol di https://console.aws.amazon.com/iam/.

2. Pilih Policies (Kebijakan). Pilih kebijakan kinesis-analytics-service-MyApplication-us-east-1 yang dibuat konsol untuk Anda di bagian sebelumnya.

3. Pilih Edit dan kemudian pilih JSONtab.

4. Tambahkan bagian yang disorot dari contoh kebijakan berikut ke kebijakan. Ganti akun sampel IDs (012345678901) dengan ID akun Anda.

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::my-bucket/kinesis-analytics-placeholder-s3-object"
               ]
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

5. Pilih Berikutnya dan kemudian pilih Simpan perubahan.

Untuk mengonfigurasi aplikasi

1. Pada MyApplicationhalaman, pilih Konfigurasi.

2. Di bagian Lokasi kode aplikasi:

   • Untuk bucket Amazon S3, pilih bucket yang sebelumnya Anda buat untuk kode aplikasi. Pilih Browse dan pilih bucket yang benar, lalu pilih Pilih. Jangan pilih nama bucket.

   • Untuk Jalur ke objek Amazon S3, masukkan managed-flink-pyflink-getting-started-1.0.0.zip.

3. Untuk izin Akses, pilih Buat/perbarui IAM peran kinesis-analytics-MyApplication-us-east-1 dengan kebijakan yang diperlukan.

4. Pindah ke properti Runtime dan pertahankan nilai default untuk semua pengaturan lainnya.

5. Pilih Tambahkan item baru dan tambahkan masing-masing parameter berikut:

   ID Grup	Kunci	Nilai
   InputStream0	stream.name	ExampleInputStream
   InputStream0	flink.stream.initpos	LATEST
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1
   kinesis.analytics.flink.run.options	python	main.py
   kinesis.analytics.flink.run.options	jarfile	lib/pyflink-dependencies.jar

6. Jangan memodifikasi bagian lain dan pilih Simpan perubahan.

Untuk menjalankan aplikasi

1. Di konsol untuk Amazon Managed Service untuk Apache Flink, pilih Aplikasi Saya dan pilih Jalankan.

2. Pada halaman berikutnya, halaman konfigurasi Pemulihan aplikasi, pilih Jalankan dengan snapshot terbaru dan kemudian pilih Jalankan.

   Status dalam Aplikasi merinci transisi dari Ready ke Starting dan kemudian ke Running saat aplikasi telah dimulai.

Untuk membuka dasbor

1. Pilih Buka dasbor Apache Flink. Dasbor terbuka di halaman baru.

2. Dalam daftar pekerjaan Runing, pilih satu pekerjaan yang dapat Anda lihat.

   catatan

   Jika Anda menyetel properti Runtime atau mengedit IAM kebijakan secara tidak benar, status aplikasi mungkin berubah menjadiRunning, tetapi dasbor Flink menunjukkan bahwa pekerjaan terus dimulai ulang. Ini adalah skenario kegagalan umum jika aplikasi salah konfigurasi atau tidak memiliki izin untuk mengakses sumber daya eksternal.

   Ketika ini terjadi, periksa tab Pengecualian di dasbor Flink untuk melihat penyebab masalah.

Untuk melihat metrik

1. Di sebelah tombol Refresh, pilih 10 detik dari daftar dropdown.

2. Saat aplikasi berjalan dan sehat, Anda dapat melihat metrik uptime terus meningkat.

3. Metrik fullrestart harus nol. Jika meningkat, konfigurasi mungkin memiliki masalah. Untuk menyelidiki masalah ini, tinjau tab Pengecualian di dasbor Flink.

4. Jumlah metrik pos pemeriksaan yang gagal harus nol dalam aplikasi yang sehat.

   catatan

   Dasbor ini menampilkan satu set metrik tetap dengan perincian 5 menit. Anda dapat membuat dasbor aplikasi khusus dengan metrik apa pun di CloudWatch dasbor.

Untuk melihat output

1. Buka konsol Kinesis di /kinesis. https://console.aws.amazon.com

2. Verifikasi bahwa Region sama dengan yang Anda gunakan untuk menjalankan tutorial ini. Secara default, itu adalah AS-Timur-1us Timur (Virginia N.). Ubah Wilayah jika perlu.

3. Pilih Aliran Data.

4. Pilih aliran yang ingin Anda amati. Untuk tutorial ini, gunakan ExampleOutputStream.

5. Pilih tab Penampil data.

6. Pilih Shard apa saja, simpan Terbaru sebagai posisi Awal, lalu pilih Dapatkan catatan. Anda mungkin melihat kesalahan “tidak ada catatan ditemukan untuk permintaan ini”. Jika demikian, pilih Coba lagi mendapatkan catatan. Catatan terbaru yang diterbitkan ke tampilan streaming.

7. Pilih nilai di kolom Data untuk memeriksa konten catatan dalam JSON format.

Untuk menghentikan aplikasi

1. Dari daftar dropdown Action, pilih Stop.

2. Status dalam Aplikasi merinci transisi dari Running keStopping, dan kemudian ke Ready saat aplikasi benar-benar dihentikan.

   catatan

   Jangan lupa juga untuk berhenti mengirim data ke input stream dari script Python atau Kinesis Data Generator.