Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris. # Memulai Amazon Managed Service untuk Apache Flink untuk Python Bagian ini memperkenalkan Anda pada konsep dasar Managed Service untuk Apache Flink menggunakan Python dan Table API. Ini menjelaskan opsi yang tersedia untuk membuat dan menguji aplikasi Anda. Ini juga memberikan petunjuk untuk menginstal alat yang diperlukan untuk menyelesaikan tutorial dalam panduan ini dan untuk membuat aplikasi pertama Anda. **Topics** + [ ## Tinjau komponen Layanan Terkelola untuk aplikasi Apache Flink ](#gs-python-table-components) + [ ## Memenuhi prasyarat ](#gs-python-prerequisites) + [ # Membuat dan menjalankan Managed Service untuk Apache Flink untuk aplikasi Python ](gs-python-createapp.md) + [ # Bersihkan AWS sumber daya ](gs-python-cleanup.md) ## Tinjau komponen Layanan Terkelola untuk aplikasi Apache Flink **catatan** Amazon Managed Service untuk Apache Flink mendukung semua [Apache](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/concepts/overview/#flinks-apis) Flink. APIs Tergantung pada API yang Anda pilih, struktur aplikasi sedikit berbeda. Salah satu pendekatan populer ketika mengembangkan aplikasi Apache Flink di Python adalah untuk mendefinisikan aliran aplikasi menggunakan SQL tertanam dalam kode Python. Ini adalah pendekatan yang kita ikuti dalam tutorial Gettgin Started berikut. Untuk memproses data, Layanan Terkelola untuk aplikasi Apache Flink Anda menggunakan skrip Python untuk menentukan aliran data yang memproses input dan menghasilkan output menggunakan runtime Apache Flink. Layanan Terkelola khas untuk aplikasi Apache Flink memiliki komponen-komponen berikut: + **Properti runtime:** Anda dapat menggunakan *properti runtime* untuk mengonfigurasi aplikasi Anda tanpa mengompilasi ulang kode aplikasi Anda. + **Sumber:** Aplikasi mengkonsumsi data dari satu atau lebih *sumber*. Sumber menggunakan [konektor](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/overview/) untuk membaca data dari sistem eksternal seperti aliran data Kinesis, atau topik MSK Amazon. Anda juga dapat menggunakan konektor khusus untuk menghasilkan data dari dalam aplikasi. *Bila Anda menggunakan SQL, aplikasi mendefinisikan sumber sebagai tabel sumber.* + **Transformasi:** Aplikasi memproses data dengan menggunakan satu atau lebih *transformasi* yang dapat menyaring, memperkaya, atau mengumpulkan data. Bila Anda menggunakan SQL, aplikasi mendefinisikan transformasi sebagai query SQL. + **Tenggelam:** Aplikasi mengirimkan data ke sumber eksternal melalui *sink*. Wastafel menggunakan [konektor](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/overview/) untuk mengirim data ke sistem eksternal seperti aliran data Kinesis, topik MSK Amazon, bucket Amazon S3, atau database relasional. Anda juga dapat menggunakan konektor khusus untuk mencetak output untuk tujuan pengembangan. Saat Anda menggunakan SQL, aplikasi mendefinisikan *sink sebagai tabel sink* tempat Anda menyisipkan hasil. Untuk informasi selengkapnya, lihat [Menulis data menggunakan sink di Managed Service untuk Apache Flink](how-sinks.md). Aplikasi Python Anda mungkin juga memerlukan dependensi eksternal, seperti pustaka Python tambahan atau konektor Flink apa pun yang digunakan aplikasi Anda. Ketika Anda mengemas aplikasi Anda, Anda harus menyertakan setiap ketergantungan yang dibutuhkan aplikasi Anda. Tutorial ini menunjukkan cara menyertakan dependensi konektor dan cara mengemas aplikasi untuk penyebaran di Amazon Managed Service untuk Apache Flink. ## Memenuhi prasyarat Untuk menyelesaikan tutorial ini, Anda harus memiliki yang berikut: + **Python 3.11** [https://docs.conda.io/en/latest/](https://docs.conda.io/en/latest/) + [Klien Git](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git) - instal klien Git jika Anda belum melakukannya. + [Java Development Kit (JDK) versi 11](https://www.oracle.com/java/technologies/downloads/#java11) - instal Java JDK 11 dan atur variabel `JAVA_HOME` lingkungan untuk menunjuk ke lokasi instalasi Anda. Jika Anda tidak memiliki JDK 11, Anda dapat menggunakan [Amazon Corretto](https://docs.aws.amazon.com/corretto)atau JDK standar pilihan kami. + Untuk memverifikasi bahwa Anda telah menginstal JDK dengan benar, jalankan perintah berikut. Outputnya akan berbeda jika Anda menggunakan JDK selain Amazon Corretto 11. Pastikan versinya 11.x. ``` $ java --version openjdk 11.0.23 2024-04-16 LTS OpenJDK Runtime Environment Corretto-11.0.23.9.1 (build 11.0.23+9-LTS) OpenJDK 64-Bit Server VM Corretto-11.0.23.9.1 (build 11.0.23+9-LTS, mixed mode) ``` + [Apache Maven](https://maven.apache.org/) - instal Apache Maven jika Anda belum melakukannya. Untuk informasi selengkapnya, lihat [Menginstal Apache Maven](https://maven.apache.org/install.html). + Untuk menguji instalasi Apache Maven Anda, gunakan perintah berikut: ``` $ mvn -version ``` **catatan** Meskipun aplikasi Anda ditulis dengan Python, Apache Flink berjalan di Java Virtual Machine (JVM). Ini mendistribusikan sebagian besar dependensi, seperti konektor Kinesis, sebagai file JAR. Untuk mengelola dependensi ini dan untuk mengemas aplikasi dalam file ZIP, gunakan [Apache](https://maven.apache.org/) Maven. Tutorial ini menjelaskan cara melakukannya. **Awas** Kami menyarankan Anda menggunakan Python 3.11 untuk pengembangan lokal. Ini adalah versi Python yang sama yang digunakan oleh Amazon Managed Service untuk Apache Flink dengan runtime Flink 1.19. Menginstal pustaka Python Flink 1.19 pada Python 3.12 mungkin gagal. Jika Anda memiliki versi Python lain yang diinstal secara default pada mesin Anda, kami sarankan Anda membuat lingkungan mandiri seperti menggunakan VirtualEnv Python 3.11. **IDE untuk pengembangan lokal** Kami menyarankan Anda menggunakan lingkungan pengembangan seperti [PyCharm](https://www.jetbrains.com/pycharm/)atau [Visual Studio Code](https://code.visualstudio.com/) untuk mengembangkan dan mengkompilasi aplikasi Anda. Kemudian, selesaikan dua langkah pertama dari[Memulai Layanan Terkelola Amazon untuk Apache Flink (API) DataStream](getting-started.md): + [Siapkan AWS akun dan buat pengguna administrator](setting-up.md) + [Mengatur AWS Command Line Interface (AWS CLI)](setup-awscli.md) Untuk memulai, lihat [Membuat aplikasi](gs-python-createapp.md). # Membuat dan menjalankan Managed Service untuk Apache Flink untuk aplikasi Python Di bagian ini, Anda membuat Layanan Terkelola untuk aplikasi Apache Flink untuk aplikasi Python dengan aliran Kinesis sebagai sumber dan wastafel. **Topics** + [ ## Buat sumber daya yang bergantung ](#gs-python-resources) + [ ## Siapkan lingkungan pengembangan lokal Anda ](#gs-python-set-up) + [ ## Unduh dan periksa kode Python streaming Apache Flink ](#gs-python-download) + [ ## Kelola dependensi JAR ](#gs-python-jar-dependencies) + [ ## Tulis catatan sampel ke aliran input ](#gs-python-sample-records) + [ ## Jalankan aplikasi Anda secara lokal ](#gs-python-run-locally) + [ ## Amati data input dan output dalam aliran Kinesis ](#gs-python-observe-input-output) + [ ## Menghentikan aplikasi Anda berjalan secara lokal ](#gs-python-stop) + [ ## Package kode aplikasi Anda ](#gs-python-package-code) + [ ## Unggah paket aplikasi ke bucket Amazon S3 ](#gs-python-upload-bucket) + [ ## Buat dan konfigurasikan Layanan Terkelola untuk aplikasi Apache Flink ](#gs-python-7) + [ ## Langkah berikutnya ](#gs-python-next-step-4) ## Buat sumber daya yang bergantung Sebelum Anda membuat Layanan Terkelola untuk Apache Flink untuk latihan ini, Anda membuat sumber daya dependen berikut: + Dua aliran Kinesis untuk input dan output. + Bucket Amazon S3 untuk menyimpan kode aplikasi. **catatan** Tutorial ini mengasumsikan bahwa Anda menerapkan aplikasi Anda di Wilayah us-east-1. Jika Anda menggunakan Wilayah lain, Anda harus menyesuaikan semua langkah yang sesuai. ### Buat dua aliran Kinesis Sebelum Anda membuat Layanan Terkelola untuk aplikasi Apache Flink untuk latihan ini, buat dua aliran data Kinesis (`ExampleInputStream`dan`ExampleOutputStream`) di Wilayah yang sama yang akan Anda gunakan untuk menyebarkan aplikasi Anda (us-east-1 dalam contoh ini). Aplikasi Anda menggunakan aliran ini untuk sumber aplikasi dan aliran tujuan. Anda dapat membuat aliran ini menggunakan konsol Amazon Kinesis atau perintah AWS CLI berikut. Untuk instruksi konsol, lihat [Membuat dan Memperbarui Aliran Data](https://docs.aws.amazon.com/kinesis/latest/dev/amazon-kinesis-streams.html) di *Panduan Developer Amazon Kinesis Data Streams*. **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 ``` 1. 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 ``` ### Buat bucket Amazon S3. Anda dapat membuat bucket Amazon S3 menggunakan konsol. Untuk petunjuk pembuatan sumber daya ini, lihat topik berikut: + [Bagaimana Cara Membuat Bucket S3?](https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket.html) di *Panduan Pengguna Layanan Penyimpanan Sederhana Amazon*. Berikan bucket Amazon S3 nama yang unik secara global, misalnya dengan menambahkan nama login Anda. **catatan** Pastikan Anda membuat bucket S3 di Region yang Anda gunakan untuk tutorial ini (us-east-1). ### Sumber daya lainnya Saat Anda membuat aplikasi, Managed Service for Apache Flink akan membuat CloudWatch resource Amazon berikut jika belum ada: + Grup log yang disebut `/AWS/KinesisAnalytics-java/`. + Aliran log yang disebut `kinesis-analytics-log-stream`. ## Siapkan lingkungan pengembangan lokal Anda Untuk pengembangan dan debugging, Anda dapat menjalankan aplikasi Python Flink di mesin Anda. Anda dapat memulai aplikasi dari baris perintah dengan `python main.py` atau dalam IDE Python pilihan Anda. **catatan** Pada mesin pengembangan Anda, Anda harus menginstal Python 3.10 atau 3.11, Java 11, Apache Maven, dan Git. Kami menyarankan Anda menggunakan IDE seperti [PyCharm](https://www.jetbrains.com/pycharm/)atau [Visual Studio Code](https://code.visualstudio.com/). Untuk memverifikasi bahwa Anda memenuhi semua prasyarat, lihat [Memenuhi prasyarat untuk menyelesaikan latihan](gs-python.md#gs-python-prerequisites) sebelum melanjutkan. ### Instal PyFlink perpustakaan Untuk mengembangkan aplikasi Anda dan menjalankannya secara lokal, Anda harus menginstal pustaka Flink Python. 1. Buat lingkungan Python mandiri VirtualEnv menggunakan, Conda, atau alat Python serupa. 1. Instal PyFlink perpustakaan di lingkungan itu. Gunakan versi runtime Apache Flink yang sama yang akan Anda gunakan di Amazon Managed Service untuk Apache Flink. Saat ini, runtime yang disarankan adalah 1.19.1. ``` $ pip install apache-flink==1.19.1 ``` 1. Pastikan bahwa lingkungan aktif ketika Anda menjalankan aplikasi Anda. Jika Anda menjalankan aplikasi di IDE, pastikan IDE menggunakan lingkungan sebagai runtime. Prosesnya tergantung pada IDE yang Anda gunakan. **catatan** Anda hanya perlu menginstal PyFlink perpustakaan. Anda **tidak** perlu menginstal cluster Apache Flink di mesin Anda. ### Otentikasi sesi Anda AWS Aplikasi ini menggunakan aliran data Kinesis untuk mempublikasikan data. Saat berjalan secara lokal, Anda harus memiliki sesi AWS otentikasi yang valid dengan izin untuk menulis ke aliran data Kinesis. Gunakan langkah-langkah berikut untuk mengautentikasi sesi Anda: 1. Jika Anda tidak memiliki AWS CLI dan profil bernama dengan kredensi valid yang dikonfigurasi, lihat[Mengatur AWS Command Line Interface (AWS CLI)](setup-awscli.md). 1. Verifikasi bahwa Anda AWS CLI telah dikonfigurasi dengan benar dan pengguna Anda memiliki izin untuk menulis ke aliran data Kinesis dengan menerbitkan catatan pengujian berikut: ``` $ aws kinesis put-record --stream-name ExampleOutputStream --data TEST --partition-key TEST ``` 1. Jika IDE Anda memiliki plugin untuk diintegrasikan AWS, Anda dapat menggunakannya untuk meneruskan kredensil ke aplikasi yang berjalan di IDE. Untuk informasi selengkapnya, lihat [AWS Toolkit for PyCharm](https://aws.amazon.com/pycharm/), [AWS Toolkit for Visual Studio](https://aws.amazon.com/visualstudiocode/) Code, [AWS dan Toolkit for](https://aws.amazon.com/intellij/) IntelliJ IDEA. ## Unduh dan periksa kode Python streaming Apache Flink Kode aplikasi Python untuk contoh ini tersedia dari. GitHub Untuk mengunduh kode aplikasi, lakukan hal berikut: 1. Kloning repositori jarak jauh menggunakan perintah berikut: ``` git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git ``` 1. Buka direktori `./python/GettingStarted` tersebut. ### Tinjau komponen aplikasi Kode aplikasi terletak di`main.py`. Kami menggunakan SQL tertanam dalam Python untuk menentukan aliran aplikasi. **catatan** Untuk pengalaman pengembang yang dioptimalkan, aplikasi ini dirancang untuk berjalan tanpa perubahan kode apa pun baik di Amazon Managed Service untuk Apache Flink maupun secara lokal, untuk pengembangan di mesin Anda. Aplikasi ini menggunakan variabel lingkungan `IS_LOCAL = true` untuk mendeteksi ketika sedang berjalan secara lokal. Anda harus mengatur variabel lingkungan `IS_LOCAL = true` baik di shell Anda atau dalam konfigurasi run IDE Anda. + Aplikasi mengatur lingkungan eksekusi dan membaca konfigurasi runtime. Untuk bekerja baik di Amazon Managed Service untuk Apache Flink dan secara lokal, aplikasi memeriksa variabel. `IS_LOCAL` + Berikut ini adalah perilaku default saat aplikasi berjalan di Amazon Managed Service untuk Apache Flink: 1. Muat dependensi yang dikemas dengan aplikasi. Untuk informasi lebih lanjut, lihat (tautan) 1. Muat konfigurasi dari properti Runtime yang Anda tentukan di Amazon Managed Service untuk aplikasi Apache Flink. Untuk informasi lebih lanjut, lihat (tautan) + Ketika aplikasi mendeteksi `IS_LOCAL = true` ketika Anda menjalankan aplikasi Anda secara lokal: 1. Memuat dependensi eksternal dari proyek. 1. Memuat konfigurasi dari `application_properties.json` file yang disertakan dalam proyek. ``` ... APPLICATION_PROPERTIES_FILE_PATH = "/etc/flink/application_properties.json" ... is_local = ( True if os.environ.get("IS_LOCAL") else False ) ... if is_local: APPLICATION_PROPERTIES_FILE_PATH = "application_properties.json" CURRENT_DIR = os.path.dirname(os.path.realpath(__file__)) table_env.get_config().get_configuration().set_string( "pipeline.jars", "file:///" + CURRENT_DIR + "/target/pyflink-dependencies.jar", ) ``` + Aplikasi mendefinisikan tabel sumber dengan `CREATE TABLE` pernyataan, menggunakan Konektor [Kinesis](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/kinesis/). Tabel ini membaca data dari aliran Kinesis masukan. Aplikasi mengambil nama aliran, Wilayah, dan posisi awal dari konfigurasi runtime. ``` table_env.execute_sql(f""" CREATE TABLE prices ( ticker VARCHAR(6), price DOUBLE, event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) PARTITIONED BY (ticker) WITH ( 'connector' = 'kinesis', 'stream' = '{input_stream_name}', 'aws.region' = '{input_stream_region}', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' ) """) ``` + Aplikasi ini juga mendefinisikan tabel wastafel menggunakan Konektor [Kinesis](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/kinesis/) dalam contoh ini. Kisah ini mengirimkan data ke aliran Kinesis keluaran. ``` table_env.execute_sql(f""" CREATE TABLE output ( ticker VARCHAR(6), price DOUBLE, event_time TIMESTAMP(3) ) PARTITIONED BY (ticker) WITH ( 'connector' = 'kinesis', 'stream' = '{output_stream_name}', 'aws.region' = '{output_stream_region}', 'sink.partitioner-field-delimiter' = ';', 'sink.batch.max-size' = '100', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' )""") ``` + Akhirnya, aplikasi mengeksekusi SQL yang tabel `INSERT INTO...` wastafel dari tabel sumber. Dalam aplikasi yang lebih kompleks, Anda mungkin memiliki langkah-langkah tambahan mengubah data sebelum menulis ke wastafel. ``` table_result = table_env.execute_sql("""INSERT INTO output SELECT ticker, price, event_time FROM prices""") ``` + Anda harus menambahkan langkah lain di akhir `main()` fungsi untuk menjalankan aplikasi secara lokal: ``` if is_local: table_result.wait() ``` Tanpa pernyataan ini, aplikasi akan segera berakhir ketika Anda menjalankannya secara lokal. Anda tidak boleh menjalankan pernyataan ini ketika Anda menjalankan aplikasi Anda di Amazon Managed Service untuk Apache Flink. ## Kelola dependensi JAR PyFlink Aplikasi biasanya membutuhkan satu atau lebih konektor. Aplikasi dalam tutorial ini menggunakan Konektor [Kinesis](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/kinesis/). Karena Apache Flink berjalan di Java JVM, konektor didistribusikan sebagai file JAR, terlepas dari apakah Anda mengimplementasikan aplikasi Anda dengan Python. Anda harus mengemas dependensi ini dengan aplikasi saat Anda menerapkannya di Amazon Managed Service untuk Apache Flink. Dalam contoh ini, kami menunjukkan bagaimana menggunakan Apache Maven untuk mengambil dependensi dan paket aplikasi untuk dijalankan pada Managed Service untuk Apache Flink. **catatan** Ada cara alternatif untuk mengambil dan mengemas dependensi. Contoh ini menunjukkan metode yang bekerja dengan benar dengan satu atau lebih konektor. Ini juga memungkinkan Anda menjalankan aplikasi secara lokal, untuk pengembangan, dan pada Managed Service untuk Apache Flink tanpa perubahan kode. ### Gunakan file pom.xml Apache Maven menggunakan `pom.xml` file untuk mengontrol dependensi dan kemasan aplikasi. Setiap dependensi JAR ditentukan dalam `pom.xml` file di blok. `...` ``` 4.0.0 ... org.apache.flink flink-connector-kinesis 4.3.0-1.19 ... ``` Untuk menemukan artefak dan versi konektor yang benar untuk digunakan, lihat[Gunakan konektor Apache Flink dengan Managed Service untuk Apache Flink](how-flink-connectors.md). Pastikan Anda merujuk ke versi Apache Flink yang Anda gunakan. Untuk contoh ini, kami menggunakan konektor Kinesis. Untuk Apache Flink 1.19, versi konektornya adalah. `4.3.0-1.19` **catatan** Jika Anda menggunakan Apache Flink 1.19, tidak ada versi konektor yang dirilis khusus untuk versi ini. Gunakan konektor yang dilepaskan untuk 1,18. ### Download dan paket dependensi Gunakan Maven untuk mengunduh dependensi yang ditentukan dalam `pom.xml` file dan mengemasnya untuk aplikasi Python Flink. 1. Arahkan ke direktori yang berisi proyek Python Getting Started yang disebut. `python/GettingStarted` 1. Jalankan perintah berikut: ``` $ mvn package ``` Maven membuat file baru bernama. `./target/pyflink-dependencies.jar` Saat Anda mengembangkan secara lokal di mesin Anda, aplikasi Python mencari file ini. **catatan** Jika Anda lupa menjalankan perintah ini, ketika Anda mencoba menjalankan aplikasi Anda, itu akan gagal dengan kesalahan: **Tidak dapat menemukan pabrik apa pun untuk pengenal “kinesis**. ## Tulis catatan sampel ke aliran input Di bagian ini, Anda akan mengirim catatan sampel ke aliran untuk aplikasi untuk diproses. Anda memiliki dua opsi untuk menghasilkan data sampel, baik menggunakan skrip Python atau [Kinesis](https://github.com/awslabs/amazon-kinesis-data-generator) Data Generator. ### Menghasilkan data sampel menggunakan skrip Python Anda dapat menggunakan skrip Python untuk mengirim catatan sampel ke aliran. **catatan** Untuk menjalankan skrip Python ini, Anda harus menggunakan Python 3.x dan menginstal pustaka SDK [AWS untuk Python](https://aws.amazon.com/developer/language/python/) (Boto). **Untuk mulai mengirim data uji ke aliran input Kinesis:** 1. Unduh skrip `stock.py` Python generator data dari repositori [generator GitHub Data](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/python/data-generator). 1. Jalankan skrip `stock.py`. ``` $ python stock.py ``` Jaga agar skrip tetap berjalan saat Anda menyelesaikan sisa tutorial. Anda sekarang dapat menjalankan aplikasi Apache Flink Anda. ### Hasilkan data sampel menggunakan Kinesis Data Generator Atau menggunakan skrip Python, Anda dapat menggunakan [Kinesis Data Generator](https://github.com/awslabs/amazon-kinesis-data-generator), juga tersedia dalam [versi yang dihosting](https://awslabs.github.io/amazon-kinesis-data-generator/web/producer.html), untuk mengirim data sampel acak ke aliran. Kinesis Data Generator berjalan di browser Anda, dan Anda tidak perlu menginstal apa pun di mesin Anda. **Untuk mengatur dan menjalankan Kinesis Data Generator:** 1. Ikuti petunjuk dalam [dokumentasi Kinesis Data Generator](https://awslabs.github.io/amazon-kinesis-data-generator/web/help.html) untuk mengatur akses ke alat. Anda akan menjalankan CloudFormation template yang mengatur pengguna dan kata sandi. 1. Akses Kinesis Data Generator melalui URL yang dihasilkan oleh template. CloudFormation Anda dapat menemukan URL di **Output** tab setelah CloudFormation template selesai. 1. Konfigurasikan generator data: + **Wilayah:** Pilih Wilayah yang Anda gunakan untuk tutorial ini: us-east-1 + **Stream/streaming pengiriman:** Pilih aliran input yang akan digunakan aplikasi: `ExampleInputStream` + **Catatan per detik:** 100 + **Rekam templat:** Salin dan tempel templat berikut: ``` { "event_time" : "{{date.now("YYYY-MM-DDTkk:mm:ss.SSSSS")}}, "ticker" : "{{random.arrayElement( ["AAPL", "AMZN", "MSFT", "INTC", "TBV"] )}}", "price" : {{random.number(100)}} } ``` 1. Uji template: Pilih **template Uji** dan verifikasi bahwa catatan yang dihasilkan mirip dengan yang berikut: ``` { "event_time" : "2024-06-12T15:08:32.04800, "ticker" : "INTC", "price" : 7 } ``` 1. Mulai generator data: Pilih **Pilih Kirim Data**. Kinesis Data Generator sekarang mengirimkan data ke file. `ExampleInputStream` ## Jalankan aplikasi Anda secara lokal Anda dapat menguji aplikasi secara lokal, berjalan dari baris perintah dengan `python main.py` atau dari IDE Anda. Untuk menjalankan aplikasi Anda secara lokal, Anda harus menginstal versi PyFlink pustaka yang benar seperti yang dijelaskan di bagian sebelumnya. Untuk informasi lebih lanjut, lihat (tautan) **catatan** Sebelum Anda melanjutkan, verifikasi bahwa aliran input dan output tersedia. Lihat [Buat dua aliran data Amazon Kinesis](get-started-exercise.md#get-started-exercise-1). Juga, verifikasi bahwa Anda memiliki izin untuk membaca dan menulis dari kedua aliran. Lihat [Otentikasi sesi Anda AWS](get-started-exercise.md#get-started-exercise-2-5). ### Impor proyek Python ke IDE Anda Untuk mulai mengerjakan aplikasi di IDE Anda, Anda harus mengimpornya sebagai proyek Python. Repositori yang Anda kloning berisi beberapa contoh. Setiap contoh adalah proyek terpisah. Untuk tutorial ini, impor konten dalam `./python/GettingStarted` subdirektori ke IDE Anda. Impor kode sebagai proyek Python yang ada. **catatan** Proses yang tepat untuk mengimpor proyek Python baru bervariasi tergantung pada IDE yang Anda gunakan. ### Periksa konfigurasi aplikasi lokal Saat berjalan secara lokal, aplikasi menggunakan konfigurasi dalam `application_properties.json` file di folder sumber daya proyek di bawah`./src/main/resources`. Anda dapat mengedit file ini untuk menggunakan nama atau Wilayah aliran Kinesis yang berbeda. ``` [ { "PropertyGroupId": "InputStream0", "PropertyMap": { "stream.name": "ExampleInputStream", "flink.stream.initpos": "LATEST", "aws.region": "us-east-1" } }, { "PropertyGroupId": "OutputStream0", "PropertyMap": { "stream.name": "ExampleOutputStream", "aws.region": "us-east-1" } } ] ``` ### Jalankan aplikasi Python Anda secara lokal Anda dapat menjalankan aplikasi Anda secara lokal, baik dari baris perintah sebagai skrip Python biasa, atau dari IDE. **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. 1. Pastikan Anda berlari `mvn package` setidaknya satu kali. 1. Atur variabel lingkungan `IS_LOCAL = true`: ``` $ export IS_LOCAL=true ``` 1. Jalankan aplikasi sebagai skrip Python biasa. ``` $python main.py ``` **Untuk menjalankan aplikasi dari dalam IDE** 1. Konfigurasikan IDE Anda untuk menjalankan `main.py` skrip dengan konfigurasi berikut: 1. Gunakan lingkungan Python mandiri seperti Conda VirtualEnv atau tempat Anda menginstal perpustakaan. PyFlink 1. Gunakan AWS kredensil untuk mengakses input dan output Kinesis aliran data. 1. Atur `IS_LOCAL = true`. 1. Proses yang tepat untuk mengatur konfigurasi run tergantung pada IDE Anda dan bervariasi. 1. Ketika Anda telah mengatur IDE Anda, jalankan skrip Python dan gunakan tooling yang disediakan oleh IDE Anda saat aplikasi sedang berjalan. ### Periksa log aplikasi secara lokal Saat berjalan secara lokal, aplikasi tidak menampilkan log apa pun di konsol, selain dari beberapa baris yang dicetak dan ditampilkan saat aplikasi dimulai. PyFlink menulis log ke file di direktori tempat pustaka Python Flink diinstal. Aplikasi mencetak lokasi log saat dimulai. Anda juga dapat menjalankan perintah berikut untuk menemukan log: ``` $ python -c "import pyflink;import os;print(os.path.dirname(os.path.abspath(pyflink.__file__))+'/log')" ``` 1. Buat daftar file di direktori logging. Anda biasanya menemukan satu `.log` file. 1. Ekor file saat aplikasi sedang berjalan:`tail -f /.log`. ## Amati data input dan output dalam aliran Kinesis Anda dapat mengamati catatan yang dikirim ke aliran input oleh (menghasilkan sampel Python) atau Kinesis Data Generator (link) dengan menggunakan **Data Viewer** di konsol Amazon Kinesis. **Untuk mengamati catatan:** [Buka konsol Kinesis di /kinesis. https://console.aws.amazon.com](https://console.aws.amazon.com/kinesis) Verifikasi bahwa Region sama dengan tempat Anda menjalankan tutorial ini, yaitu us-east-1 US East (Virginia N.) secara default. Ubah Wilayah jika tidak cocok. Pilih **Aliran Data**. Pilih aliran yang ingin Anda amati, salah satu `ExampleInputStream` atau `ExampleOutputStream.` Pilih tab **Penampil data**. 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. Pilih nilai di kolom Data untuk memeriksa konten catatan dalam format JSON. ## Menghentikan aplikasi Anda berjalan secara lokal Hentikan aplikasi yang berjalan di IDE Anda. IDE biasanya menyediakan opsi “berhenti”. Lokasi dan metode yang tepat tergantung pada IDE. ## Package kode aplikasi Anda Di bagian ini, Anda menggunakan Apache Maven untuk mengemas kode aplikasi dan semua dependensi yang diperlukan dalam file.zip. Jalankan perintah paket Maven lagi: ``` $ mvn package ``` Perintah ini menghasilkan file`target/managed-flink-pyflink-getting-started-1.0.0.zip`. ## Unggah paket aplikasi ke bucket Amazon S3 Di bagian ini, Anda mengunggah file.zip yang Anda buat di bagian sebelumnya ke bucket Amazon Simple Storage Service (Amazon S3) Storage Service (Amazon S3) yang Anda buat di awal tutorial ini. Jika Anda belum menyelesaikan langkah ini, lihat (tautan). **Untuk mengunggah file JAR kode aplikasi** 1. Buka konsol Amazon S3 di. [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/) 1. Pilih bucket yang sebelumnya Anda buat untuk kode aplikasi. 1. Pilih **Unggah**. 1. Pilih **Tambahkan file**. 1. Arahkan ke file.zip yang dihasilkan pada langkah sebelumnya:`target/managed-flink-pyflink-getting-started-1.0.0.zip`. 1. Pilih **Unggah** tanpa mengubah pengaturan lainnya. ## Buat dan konfigurasikan Layanan Terkelola untuk aplikasi Apache Flink Anda dapat membuat dan mengkonfigurasi Layanan Terkelola untuk aplikasi Apache Flink menggunakan konsol atau aplikasi. AWS CLI Untuk tutorial ini, kita akan menggunakan konsol. ### Buat aplikasi 1. Masuk ke Konsol Manajemen AWS, dan buka konsol Amazon MSF di /flink. https://console.aws.amazon.com 1. Verifikasi bahwa Wilayah yang benar dipilih: US East (Virginia N.) us-east-1. 1. 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. 1. 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/perbarui peran IAM kinesis-analytics-MyApplication-us -east-1** dengan kebijakan yang diperlukan. + Untuk **Template untuk pengaturan aplikasi**: + Untuk **Template**, pilih **Development**. + Pilih **Buat aplikasi streaming**. **catatan** Saat membuat aplikasi Managed Service for Apache Flink menggunakan konsol, Anda memiliki opsi untuk membuat peran dan kebijakan IAM untuk aplikasi Anda. Aplikasi Anda menggunakan peran dan kebijakan ini untuk mengakses sumber daya dependen. Sumber daya IAM ini diberi nama menggunakan nama aplikasi dan Wilayah sebagai berikut: Kebijakan: `kinesis-analytics-service-MyApplication-us-west-2` Peran: `kinesisanalytics-MyApplication-us-west-2` *Amazon Managed Service untuk Apache Flink sebelumnya dikenal sebagai Kinesis Data Analytics.* Nama sumber daya yang dihasilkan secara otomatis diawali dengan `kinesis-analytics` kompatibilitas mundur. ### Edit kebijakan IAM Edit kebijakan IAM untuk menambahkan izin agar dapat mengakses bucket Amazon S3. **Untuk mengedit kebijakan IAM agar dapat menambahkan izin bucket S3** 1. Buka konsol IAM di [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/). 1. Pilih **Policies** (Kebijakan). Pilih kebijakan **`kinesis-analytics-service-MyApplication-us-east-1`** yang dibuat konsol untuk Anda di bagian sebelumnya. 1. Pilih **Edit** dan kemudian pilih tab **JSON**. 1. Tambahkan bagian yang disorot dari contoh kebijakan berikut ke kebijakan. Ganti akun sampel IDs (*012345678901*) dengan ID akun Anda. ------ #### [ JSON ] **** ``` { "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" } ] } ``` ------ 1. Pilih **Berikutnya** dan kemudian pilih **Simpan perubahan**. ### Konfigurasikan aplikasi Edit konfigurasi aplikasi untuk mengatur artefak kode aplikasi. **Untuk mengonfigurasi aplikasi** 1. Pada **MyApplication**halaman, pilih **Konfigurasi**. 1. 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**. 1. Untuk **izin Akses**, pilih **Buat/perbarui peran IAM `kinesis-analytics-MyApplication-us-east-1` dengan kebijakan yang diperlukan**. 1. Pindah ke **properti Runtime** dan pertahankan nilai default untuk semua pengaturan lainnya. 1. Pilih **Tambahkan item baru** dan tambahkan masing-masing parameter berikut: [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/managed-flink/latest/java/gs-python-createapp.html) 1. Jangan memodifikasi bagian lain dan pilih **Simpan perubahan**. **catatan** Saat Anda memilih untuk mengaktifkan CloudWatch pencatatan Amazon, Layanan Terkelola untuk Apache Flink membuat grup log dan aliran log untuk Anda. Nama-nama sumber daya ini adalah sebagai berikut: Grup log: `/aws/kinesis-analytics/MyApplication` Aliran log: `kinesis-analytics-log-stream` ### Jalankan aplikasi Aplikasi sekarang dikonfigurasi dan siap dijalankan. **Untuk menjalankan aplikasi** 1. **Di konsol untuk Amazon Managed Service untuk Apache Flink, pilih **Aplikasi Saya** dan pilih Jalankan.** 1. 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. Saat aplikasi dalam `Running` status, Anda sekarang dapat membuka dasbor Flink. **Untuk membuka dasbor** 1. Pilih **Buka dasbor Apache Flink**. Dasbor terbuka di halaman baru. 1. Dalam daftar **pekerjaan Runing**, pilih satu pekerjaan yang dapat Anda lihat. **catatan** Jika Anda menyetel properti Runtime atau mengedit kebijakan IAM secara tidak benar, status aplikasi mungkin berubah menjadi`Running`, 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. ### Amati metrik aplikasi yang sedang berjalan Pada **MyApplication**halaman, di bagian ** CloudWatch metrik Amazon**, Anda dapat melihat beberapa metrik dasar dari aplikasi yang sedang berjalan. **Untuk melihat metrik** 1. Di sebelah tombol **Refresh**, pilih **10 detik** dari daftar dropdown. 1. Saat aplikasi berjalan dan sehat, Anda dapat melihat metrik **uptime** terus meningkat. 1. Metrik **fullrestart** harus nol. Jika meningkat, konfigurasi mungkin memiliki masalah. Untuk menyelidiki masalah ini, tinjau tab **Pengecualian** di dasbor Flink. 1. **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. ### Amati data keluaran dalam aliran Kinesis Pastikan Anda masih mempublikasikan data ke input, baik menggunakan script Python atau Kinesis Data Generator. Anda sekarang dapat mengamati output dari aplikasi yang berjalan pada Managed Service untuk Apache Flink dengan menggunakan Data Viewer di [https://console.aws.amazon.com/kinesis/](https://console.aws.amazon.com/kinesis/), mirip dengan apa yang sudah Anda lakukan sebelumnya. **Untuk melihat output** 1. [Buka konsol Kinesis di /kinesis. https://console.aws.amazon.com](https://console.aws.amazon.com/kinesis) 1. 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. 1. Pilih **Aliran Data**. 1. Pilih aliran yang ingin Anda amati. Untuk tutorial ini, gunakan `ExampleOutputStream`. 1. Pilih tab **Penampil data**. 1. 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. 1. Pilih nilai di kolom Data untuk memeriksa konten catatan dalam format JSON. ### Hentikan aplikasi Untuk menghentikan aplikasi, buka halaman konsol dari aplikasi Managed Service for Apache Flink bernama. `MyApplication` **Untuk menghentikan aplikasi** 1. **Dari daftar dropdown **Action**, pilih Stop.** 1. **Status** dalam **Aplikasi merinci** transisi dari `Running` ke`Stopping`, dan kemudian ke `Ready` saat aplikasi benar-benar dihentikan. **catatan** Jangan lupa juga berhenti mengirim data ke input stream dari script Python atau Kinesis Data Generator. ## Langkah berikutnya [Bersihkan AWS sumber daya](gs-python-cleanup.md) # Bersihkan AWS sumber daya Bagian ini mencakup prosedur untuk membersihkan AWS sumber daya yang dibuat dalam tutorial Memulai (Python). **Topics** + [ ## Hapus Layanan Terkelola Anda untuk aplikasi Apache Flink ](#gs-python-cleanup-app) + [ ## Hapus aliran data Kinesis ](#gs-python-cleanup-msk) + [ ## Hapus objek dan bucket Amazon S3 Anda ](#gs-python-cleanup-s3) + [ ## Hapus sumber daya IAM Anda ](#gs-python-cleanup-iam) + [ ## Hapus CloudWatch sumber daya Anda ](#gs-python-cleanup-cw) ## Hapus Layanan Terkelola Anda untuk aplikasi Apache Flink Gunakan prosedur berikut untuk menghapus aplikasi. **Untuk menghapus aplikasi** 1. [Buka konsol Kinesis di /kinesis. https://console.aws.amazon.com](https://console.aws.amazon.com/kinesis) 1. Di panel Managed Service for Apache Flink, pilih. **MyApplication** 1. Dari daftar dropdown **Tindakan**, pilih **Hapus** dan kemudian konfirmasikan penghapusan. ## Hapus aliran data Kinesis 1. Masuk ke Konsol Manajemen AWS, dan buka konsol Amazon MSF di /flink. https://console.aws.amazon.com 1. Pilih **Aliran data**. 1. Pilih dua aliran yang Anda buat, `ExampleInputStream` dan`ExampleOutputStream`. 1. Dari daftar dropdown **Tindakan**, pilih **Hapus**, lalu konfirmasikan penghapusan. ## Hapus objek dan bucket Amazon S3 Anda Gunakan prosedur berikut untuk menghapus objek dan bucket S3 Anda. **Untuk menghapus objek dari bucket S3** 1. Buka konsol Amazon S3 di. [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/) 1. Pilih bucket S3 yang Anda buat untuk artefak aplikasi. 1. Pilih artefak aplikasi yang Anda unggah, bernama. `amazon-msf-java-stream-app-1.0.jar` 1. Pilih **Hapus** dan konfirmasikan penghapusan. **Untuk menghapus bucket S3** 1. Buka konsol Amazon S3 di. [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/) 1. Pilih ember yang Anda buat untuk artefak. 1. Pilih **Hapus** dan konfirmasikan penghapusan. **catatan** Bucket S3 harus kosong untuk menghapusnya. ## Hapus sumber daya IAM Anda Gunakan prosedur berikut untuk menghapus sumber daya IAM Anda. **Untuk menghapus sumber daya IAM Anda** 1. Buka konsol IAM di [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/). 1. Di bilah navigasi, pilih **Policies** (Kebijakan). 1. Di kontrol filter, masukkan **kinesis**. 1. Pilih kebijakan **kinesis-analytics-service- MyApplication -us-east-1**. 1. Pilih **Policy Actions** (Tindakan Kebijakan), lalu pilih **Delete** (Hapus). 1. Di bilah navigasi, pilih **Roles** (Peran). 1. Pilih peran **kinesis-analytics- -us-east-1**. MyApplication 1. Pilih **Delete role** (Hapus peran), lalu konfirmasi penghapusan. ## Hapus CloudWatch sumber daya Anda Gunakan prosedur berikut untuk menghapus CloudWatch sumber daya Anda. **Untuk menghapus CloudWatch sumber daya Anda** 1. Buka CloudWatch konsol di [https://console.aws.amazon.com/cloudwatch/](https://console.aws.amazon.com/cloudwatch/). 1. Di bilah navigasi, pilih **Logs**. 1. Pilih grup**/aws/kinesis-analytics/MyApplication**log. 1. Pilih **Delete Log Group** (Hapus Grup Log), lalu konfirmasi penghapusan.