Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Layanan Terkelola untuk Apache Flink: Cara kerjanya
Managed Service for Apache Flink adalah layanan Amazon yang dikelola sepenuhnya yang memungkinkan Anda menggunakan aplikasi Apache Flink untuk memproses data streaming. Pertama, Anda memprogram aplikasi Apache Flink Anda, dan kemudian Anda membuat Layanan Terkelola untuk aplikasi Apache Flink Anda.
## Program aplikasi Apache Flink Anda
Aplikasi Apache Flink adalah aplikasi Java atau Scala yang dibuat dengan kerangka kerja Apache Flink. Anda menulis dan membangun aplikasi Apache Flink Anda secara lokal.
Aplikasi terutama menggunakan [DataStream API](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/datastream_api.html) atau [Table API](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/table/). Apache Flink lainnya juga APIs tersedia untuk Anda gunakan, tetapi mereka kurang umum digunakan dalam membangun aplikasi streaming.
Fitur keduanya APIs adalah sebagai berikut:
### DataStream API
Model pemrograman DataStream API Apache Flink didasarkan pada dua komponen:
+ **Aliran data:** Representasi terstruktur dari aliran catatan data yang berkelanjutan.
+ **Operator transformasi:** Membawa satu atau beberapa aliran data sebagai input, dan menghasilkan satu atau beberapa aliran data sebagai output.
Aplikasi yang dibuat dengan DataStream API melakukan hal berikut:
+ Baca data dari Sumber Data (seperti aliran Kinesis atau topik Amazon MSK).
+ Terapkan transformasi ke data, seperti penyaringan, agregasi, atau pengayaan.
+ Tulis data yang diubah ke Sink Data.
Aplikasi yang menggunakan DataStream API dapat ditulis dalam Java atau Scala, dan dapat dibaca dari aliran data Kinesis, topik MSK Amazon, atau sumber kustom.
Aplikasi Anda memproses data menggunakan *konektor*. Apache Flink menggunakan tipe konektor berikut:
+ **Source** (Sumber) : Konektor yang digunakan untuk membaca data eksternal.
+ **Sink**: Konektor yang digunakan untuk menulis ke lokasi eksternal.
+ **Operator**: Konektor yang digunakan untuk memproses data dalam aplikasi.
Aplikasi yang khas terdiri dari setidaknya satu aliran data dengan sumber, aliran data dengan satu atau beberapa operator, dan setidaknya satu data sink.
Untuk informasi selengkapnya tentang penggunaan DataStream API, lihat[Tinjau komponen DataStream API](how-datastream.md).
### Tabel API
Model pemrograman API Tabel Apache Flink didasarkan pada komponen berikut:
+ **Lingkungan Tabel:** Antarmuka untuk data yang mendasari yang Anda gunakan untuk membuat dan meng-host satu atau beberapa tabel.
+ **Tabel:** Objek yang menyediakan akses ke tabel atau tampilan SQL.
+ **Sumber Tabel:** Digunakan untuk membaca data dari sumber eksternal, seperti topik Amazon MSK.
+ **Fungsi Tabel:** Kueri SQL atau panggilan API yang digunakan untuk mengubah data.
+ **Sink Tabel:** Digunakan untuk menulis data ke lokasi eksternal, seperti bucket Amazon S3.
Aplikasi yang dibuat dengan API Tabel melakukan hal berikut:
+ Buat `TableEnvironment` dengan menghubungkan ke `Table Source`.
+ Buat tabel di `TableEnvironment` menggunakan kueri SQL atau fungsi API Tabel.
+ Jalankan kueri pada tabel menggunakan API Tabel atau SQL
+ Terapkan transformasi pada hasil kueri menggunakan Fungsi Tabel atau kueri SQL.
+ Tulis hasil kueri atau fungsi ke `Table Sink`.
Aplikasi yang menggunakan API Tabel dapat ditulis di Java atau Scala, dan dapat mengkueri data menggunakan panggilan API atau kueri SQL.
Untuk informasi selengkapnya tentang penggunaan API Tabel, lihat [Tinjau komponen API Tabel](how-table.md).
## Buat Layanan Terkelola Anda untuk aplikasi Apache Flink
Managed Service for Apache Flink adalah AWS layanan yang menciptakan lingkungan untuk hosting aplikasi Apache Flink Anda dan menyediakannya dengan pengaturan berikut:
+ **[Gunakan properti runtime](how-properties.md): ** Parameter yang dapat Anda berikan ke aplikasi Anda. Anda dapat mengubah parameter ini tanpa mengompilasi ulang kode aplikasi Anda.
+ **[Menerapkan toleransi kesalahan](how-fault.md)**: Cara aplikasi Anda pulih dari gangguan dan mulai ulang.
+ **[Pencatatan dan pemantauan di Amazon Managed Service untuk Apache Flink](monitoring-overview.md)**: Bagaimana aplikasi Anda mencatat peristiwa ke CloudWatch Log.
+ **[Menerapkan penskalaan aplikasi](how-scaling.md)**: Cara aplikasi Anda menyediakan sumber daya komputasi.
Anda membuat Layanan Terkelola untuk aplikasi Apache Flink menggunakan konsol atau aplikasi. AWS CLI Untuk mulai membuat Layanan Terkelola untuk aplikasi Apache Flink, lihat. [Tutorial: Mulai menggunakan DataStream API di Managed Service untuk Apache Flink](getting-started.md)
# Buat Layanan Terkelola untuk aplikasi Apache Flink
Topik ini berisi informasi tentang membuat Layanan Terkelola untuk aplikasi Apache Flink.
**Topics**
+ [Bangun Layanan Terkelola Anda untuk kode aplikasi Apache Flink](#how-creating-apps-building)
+ [Buat Layanan Terkelola Anda untuk aplikasi Apache Flink](#how-creating-apps-creating)
+ [Gunakan kunci yang dikelola pelanggan](#how-creating-apps-use-cmk)
+ [Mulai Layanan Terkelola Anda untuk aplikasi Apache Flink](#how-creating-apps-starting)
+ [Verifikasi Layanan Terkelola Anda untuk aplikasi Apache Flink](#how-creating-apps-verifying)
+ [Aktifkan rollback sistem untuk Layanan Terkelola Anda untuk aplikasi Apache Flink](how-system-rollbacks.md)
## Bangun Layanan Terkelola Anda untuk kode aplikasi Apache Flink
Bagian ini menjelaskan komponen yang Anda gunakan untuk membangun kode aplikasi untuk Layanan Terkelola untuk aplikasi Apache Flink Anda.
Sebaiknya gunakan versi terbaru Apache Flink yang didukung untuk kode aplikasi Anda. Untuk informasi tentang memutakhirkan Layanan Terkelola untuk aplikasi Apache Flink, lihat. [Gunakan upgrade versi di tempat untuk Apache Flink](how-in-place-version-upgrades.md)
Anda membangun kode aplikasi Anda menggunakan [Apache Maven](https://maven.apache.org/). Proyek Apache Maven menggunakan file `pom.xml` untuk menentukan versi komponen yang digunakan.
**catatan**
Layanan Terkelola untuk Apache Flink mendukung file JAR hingga ukuran 512 MB. Jika Anda menggunakan file JAR lebih besar dari ini, aplikasi Anda akan gagal dimulai.
Aplikasi sekarang dapat menggunakan Java API dari versi Scala apa pun. Anda harus menggabungkan pustaka standar Scala pilihan Anda ke dalam aplikasi Scala Anda.
Untuk informasi tentang membuat Layanan Terkelola untuk aplikasi Apache Flink yang menggunakan **Apache Beam**, lihat. [Gunakan Apache Beam dengan Managed Service untuk aplikasi Apache Flink](how-creating-apps-beam.md)
### Tentukan versi Apache Flink aplikasi Anda
Saat menggunakan Managed Service for Apache Flink Runtime versi 1.1.0 dan yang lebih baru, Anda menentukan versi Apache Flink yang digunakan aplikasi Anda saat Anda mengkompilasi aplikasi Anda. Anda memberikan versi Apache Flink dengan parameter. `-Dflink.version` Misalnya, jika Anda menggunakan Apache Flink 2.2.0, berikan yang berikut ini:
```
mvn package -Dflink.version=2.2.0
```
Untuk membangun aplikasi dengan versi Apache Flink sebelumnya, lihat. [Versi sebelumnya](earlier.md)
## Buat Layanan Terkelola Anda untuk aplikasi Apache Flink
Setelah membuat kode aplikasi, lakukan hal berikut untuk membuat aplikasi Managed Service for Apache Flink (Amazon MSF):
+ **Unggah kode Aplikasi Anda**: Unggah kode aplikasi Anda ke bucket Amazon S3. Anda menentukan nama bucket S3 dan nama objek kode aplikasi Anda ketika membuat aplikasi Anda. Untuk tutorial yang menunjukkan cara mengunggah kode aplikasi Anda, lihat [Tutorial: Mulai menggunakan DataStream API di Managed Service untuk Apache Flink](getting-started.md) tutorialnya.
+ **Buat Layanan Terkelola untuk aplikasi Apache Flink** Anda: Gunakan salah satu metode berikut untuk membuat aplikasi MSF Amazon Anda:
**catatan**
Amazon MSF mengenkripsi aplikasi Anda secara default menggunakan. Kunci milik AWS Anda juga dapat membuat aplikasi baru menggunakan kunci terkelola AWS KMS pelanggan (CMKs) untuk membuat, memiliki, dan mengelola kunci Anda sendiri. Untuk informasi tentang CMKs, lihat[Manajemen kunci di Amazon Managed Service untuk Apache Flink](key-management-flink.md).
+ **Buat aplikasi Amazon MSF Anda menggunakan AWS konsol:** Anda dapat membuat dan mengonfigurasi aplikasi Anda menggunakan AWS konsol.
Saat Anda membuat aplikasi menggunakan konsol, sumber daya dependen aplikasi Anda (seperti aliran CloudWatch Log, peran IAM, dan kebijakan IAM) akan dibuat untuk Anda.
Saat Anda membuat aplikasi menggunakan konsol, Anda menentukan versi Apache Flink yang digunakan aplikasi Anda dengan memilihnya dari pull-down pada halaman **Managed Service for Apache Flink** - Create application.
Untuk tutorial tentang cara menggunakan konsol untuk membuat aplikasi, lihat [Tutorial: Mulai menggunakan DataStream API di Managed Service untuk Apache Flink](getting-started.md) tutorialnya.
+ **Buat aplikasi Amazon MSF Anda menggunakan AWS CLI**: Anda dapat membuat dan mengkonfigurasi aplikasi Anda menggunakan CLI. AWS
Saat Anda membuat aplikasi menggunakan CLI, Anda juga harus membuat sumber daya dependen aplikasi Anda (seperti aliran CloudWatch Log, peran IAM, dan kebijakan IAM) secara manual.
Ketika Anda membuat aplikasi Anda menggunakan CLI, Anda menentukan versi Apache Flink yang digunakan aplikasi Anda menggunakan parameter `RuntimeEnvironment` dari tindakan `CreateApplication`.
**catatan**
Anda dapat mengubah `RuntimeEnvironment` aplikasi yang ada. Untuk mempelajari caranya, lihat [Gunakan upgrade versi di tempat untuk Apache Flink](how-in-place-version-upgrades.md).
## Gunakan kunci yang dikelola pelanggan
Di Amazon MSF, kunci yang dikelola pelanggan (CMKs) adalah fitur yang dapat digunakan untuk mengenkripsi data aplikasi Anda dengan kunci yang Anda buat, miliki, dan kelola pada AWS Key Management Service ()AWS KMS. Untuk aplikasi Amazon MSF, ini berarti semua data yang tunduk pada [pos pemeriksaan](how-fault.md) Flink atau [snapshot](how-snapshots.md) dienkripsi dengan CMK yang Anda tentukan untuk aplikasi itu.
Untuk menggunakan CMK dengan aplikasi Anda, Anda harus terlebih dahulu [membuat aplikasi baru Anda](#how-creating-apps-creating), dan kemudian menerapkan CMK. Untuk informasi selengkapnya tentang penggunaan CMKs, lihat[Manajemen kunci di Amazon Managed Service untuk Apache Flink](key-management-flink.md).
## Mulai Layanan Terkelola Anda untuk aplikasi Apache Flink
Setelah Anda membangun kode aplikasi Anda, mengunggahnya ke S3, dan membuat Layanan Terkelola untuk aplikasi Apache Flink, Anda kemudian memulai aplikasi Anda. Memulai Layanan Terkelola untuk aplikasi Apache Flink biasanya memakan waktu beberapa menit.
Gunakan salah satu metode berikut untuk memulai aplikasi Anda:
+ **Mulai Layanan Terkelola untuk aplikasi Apache Flink menggunakan AWS konsol:** Anda dapat menjalankan aplikasi Anda dengan memilih **Jalankan** pada halaman aplikasi Anda di AWS konsol.
+ **Mulai Layanan Terkelola untuk aplikasi Apache Flink menggunakan AWS API:** Anda dapat menjalankan aplikasi Anda menggunakan tindakan. [StartApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StartApplication.html)
## Verifikasi Layanan Terkelola Anda untuk aplikasi Apache Flink
Anda dapat memverifikasi bahwa aplikasi Anda bekerja dengan cara berikut:
+ **Menggunakan CloudWatch Log:** Anda dapat menggunakan Wawasan CloudWatch CloudWatch Log dan Log untuk memverifikasi bahwa aplikasi Anda berjalan dengan benar. Untuk informasi tentang menggunakan CloudWatch Log dengan Layanan Terkelola untuk aplikasi Apache Flink, lihat. [Pencatatan dan pemantauan di Amazon Managed Service untuk Apache Flink](monitoring-overview.md)
+ **Menggunakan CloudWatch Metrik:** Anda dapat menggunakan CloudWatch Metrik untuk memantau aktivitas aplikasi, atau aktivitas dalam sumber daya yang digunakan aplikasi untuk input atau output (seperti aliran Kinesis, aliran Firehose, atau bucket Amazon S3.) Untuk informasi selengkapnya tentang CloudWatch metrik, lihat [Bekerja dengan Metrik](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/working_with_metrics.html) di CloudWatch Panduan Pengguna Amazon.
+ **Pemantauan Lokasi Output:** Jika aplikasi Anda menulis output ke lokasi (seperti bucket atau basis data Amazon S3), Anda dapat memantau lokasi tersebut untuk data tertulis.
# Aktifkan rollback sistem untuk Layanan Terkelola Anda untuk aplikasi Apache Flink
Dengan kemampuan system-rollback, Anda dapat mencapai ketersediaan yang lebih tinggi dari aplikasi Apache Flink yang sedang berjalan di Amazon Managed Service untuk Apache Flink. Memilih ke konfigurasi ini memungkinkan layanan untuk secara otomatis mengembalikan aplikasi ke versi yang berjalan sebelumnya ketika tindakan seperti `UpdateApplication` atau `autoscaling` berjalan ke kode atau konfigurasi bug.
**catatan**
Untuk menggunakan fitur rollback sistem, Anda harus ikut serta dengan memperbarui aplikasi Anda. Aplikasi yang ada tidak akan secara otomatis menggunakan rollback sistem secara default.
## Cara kerjanya
Saat Anda memulai operasi aplikasi, seperti tindakan pembaruan atau penskalaan, Amazon Managed Service untuk Apache Flink pertama kali mencoba menjalankan operasi tersebut. Jika mendeteksi masalah yang mencegah operasi berhasil, seperti bug kode atau izin yang tidak memadai, layanan secara otomatis memulai operasi. `RollbackApplication`
Rollback mencoba mengembalikan aplikasi ke versi sebelumnya yang berhasil berjalan, bersama dengan status aplikasi terkait. Jika rollback berhasil, aplikasi Anda terus memproses data dengan downtime minimal menggunakan versi sebelumnya. Jika rollback otomatis juga gagal, Amazon Managed Service untuk Apache Flink mentransisikan aplikasi ke `READY` status, sehingga Anda dapat mengambil tindakan lebih lanjut, termasuk memperbaiki kesalahan dan mencoba kembali operasi.
Anda harus memilih untuk menggunakan rollback sistem otomatis. Anda dapat mengaktifkannya menggunakan konsol atau API untuk semua operasi pada aplikasi Anda mulai saat ini.
Contoh permintaan berikut untuk `UpdateApplication` tindakan memungkinkan rollback sistem untuk aplikasi:
```
{
"ApplicationName": "MyApplication",
"CurrentApplicationVersionId": 1,
"ApplicationConfigurationUpdate": {
"ApplicationSystemRollbackConfigurationUpdate": {
"RollbackEnabledUpdate": "true"
}
}
}
```
## Tinjau skenario umum untuk rollback sistem otomatis
Skenario berikut menggambarkan di mana kemunduran sistem otomatis bermanfaat:
+ **Pembaruan aplikasi:** Jika Anda memperbarui aplikasi Anda dengan kode baru yang memiliki bug saat menginisialisasi pekerjaan Flink melalui metode utama, rollback otomatis memungkinkan versi kerja sebelumnya dipulihkan. Skenario pembaruan lain di mana rollback sistem membantu meliputi:
+ [Jika aplikasi Anda diperbarui untuk berjalan dengan paralelisme yang lebih tinggi dari MaxParallelism.](https://docs.aws.amazon.com/managed-flink/latest/java/how-scaling.html#how-scaling-auto)
+ Jika aplikasi Anda diperbarui untuk berjalan dengan subnet yang salah untuk aplikasi VPC yang mengakibatkan kegagalan selama startup pekerjaan Flink.
+ **Peningkatan versi Flink:** Saat Anda meningkatkan ke versi Apache Flink baru dan aplikasi yang ditingkatkan mengalami masalah kompatibilitas snapshot, rollback sistem memungkinkan Anda kembali ke versi Flink sebelumnya secara otomatis.
+ **AutoScaling:** Saat aplikasi ditingkatkan tetapi mengalami masalah pemulihan dari savepoint, karena ketidakcocokan operator antara snapshot dan grafik pekerjaan Flink.
## Gunakan operasi APIs untuk rollback sistem
Untuk memberikan visibilitas yang lebih baik, Amazon Managed Service untuk Apache Flink memiliki dua yang APIs terkait dengan operasi aplikasi yang dapat membantu Anda melacak kegagalan dan rollback sistem terkait.
`ListApplicationOperations`
API ini mencantumkan semua operasi yang dilakukan pada aplikasi, termasuk`UpdateApplication`,`Maintenance`,`RollbackApplication`, dan lainnya dalam urutan kronologis terbalik. Contoh permintaan berikut untuk `ListApplicationOperations` tindakan mencantumkan 10 operasi aplikasi pertama untuk aplikasi:
```
{
"ApplicationName": "MyApplication",
"Limit": 10
}
```
Contoh permintaan bantuan berikut `ListApplicationOperations` ini memfilter daftar ke pembaruan sebelumnya pada aplikasi:
```
{
"ApplicationName": "MyApplication",
"operation": "UpdateApplication"
}
```
`DescribeApplicationOperation`
API ini memberikan informasi terperinci tentang operasi tertentu yang dicantumkan oleh`ListApplicationOperations`, termasuk alasan kegagalan, jika berlaku. Contoh permintaan berikut untuk `DescribeApplicationOperation` tindakan mencantumkan rincian untuk operasi aplikasi tertentu:
```
{
"ApplicationName": "MyApplication",
"OperationId": "xyzoperation"
}
```
Untuk informasi pemecahan masalah, lihat [Praktik terbaik rollback sistem](troubleshooting-system-rollback.md).
# Jalankan Layanan Terkelola untuk aplikasi Apache Flink
Topik ini berisi informasi tentang menjalankan Managed Service untuk Apache Flink.
Saat Anda menjalankan aplikasi Managed Service for Apache Flink, layanan akan membuat pekerjaan Apache Flink. Pekerjaan Apache Flink adalah siklus hidup eksekusi Layanan Terkelola untuk aplikasi Apache Flink Anda. Eksekusi tugas, dan sumber daya yang digunakannya, dikelola oleh Manajer Tugas. Manajer Tugas memisahkan eksekusi aplikasi ke dalam tugas-tugas. Setiap tugas dikelola oleh Manajer Tugas. Ketika Anda memantau performa aplikasi, Anda dapat memeriksa performa masing-masing Manajer Tugas, atau Manajer Tugas secara keseluruhan.
Untuk informasi tentang pekerjaan Apache Flink, lihat [Pekerjaan dan Penjadwalan](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/internals/job_scheduling/) di Dokumentasi Apache Flink.
## Identifikasi lamaran dan status pekerjaan
Aplikasi Anda dan tugas aplikasi memiliki status eksekusi saat ini:
+ **Status aplikasi:** Aplikasi Anda memiliki status saat ini yang menggambarkan fase eksekusi. Status aplikasi mencakup hal-hal berikut:
+ **Status aplikasi stabil:** Aplikasi Anda biasanya tetap berada dalam status ini hingga Anda membuat perubahan status:
+ **READY** (SIAP): Aplikasi baru atau yang dihentikan berada dalam status READY (SIAP) hingga Anda menjalankannya.
+ **RUNNING** (BERJALAN): Aplikasi yang telah berhasil dimulai berada dalam status RUNNING (BERJALAN).
+ **Status aplikasi sementara:** Aplikasi dalam status ini biasanya dalam proses transisi ke status lain. Jika aplikasi tetap dalam status sementara untuk jangka waktu yang lama, Anda dapat menghentikan aplikasi menggunakan [StopApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StopApplication.html)tindakan dengan `Force` parameter yang disetel ke. `true` Status ini mencakup hal berikut:
+ `STARTING:`Terjadi setelah [StartApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StartApplication.html)tindakan. Aplikasi ini bertransisi dari status `READY` ke `RUNNING`.
+ `STOPPING:`Terjadi setelah [StopApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StopApplication.html)tindakan. Aplikasi ini bertransisi dari status `RUNNING` ke `READY`.
+ `DELETING:`Terjadi setelah [DeleteApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DeleteApplication.html)tindakan. Aplikasi sedang dalam proses penghapusan.
+ `UPDATING:`Terjadi setelah [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)tindakan. Aplikasi memperbarui, dan akan bertransisi kembali ke status `RUNNING` atau `READY`.
+ `AUTOSCALING:`Aplikasi ini memiliki `AutoScalingEnabled` properti [ ParallelismConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ParallelismConfiguration.html)set ke`true`, dan layanan meningkatkan paralelisme aplikasi. Ketika aplikasi dalam status ini, satu-satunya tindakan API valid yang dapat Anda gunakan adalah [StopApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StopApplication.html)tindakan dengan `Force` parameter yang disetel ke`true`. Untuk informasi tentang penskalaan otomatis, lihat [Gunakan penskalaan otomatis di Managed Service untuk Apache Flink](how-scaling-auto.md).
+ `FORCE_STOPPING:`Terjadi setelah [StopApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StopApplication.html)tindakan dipanggil dengan `Force` parameter diatur ke`true`. Aplikasi sedang dalam proses penghentian paksa. Aplikasi bertransisi dari status `STARTING`, `UPDATING`, `STOPPING`, atau `AUTOSCALING` ke status `READY`.
+ `ROLLING_BACK:`Terjadi setelah [RollbackApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_RollbackApplication.html)tindakan dipanggil. Aplikasi sedang dalam proses dikembalikan ke versi sebelumnya. Aplikasi bertransisi dari status `UPDATING` atau `AUTOSCALING` ke status `RUNNING`.
+ `MAINTENANCE:`Terjadi saat Managed Service for Apache Flink menerapkan patch ke aplikasi Anda. Untuk informasi selengkapnya, lihat [Mengelola tugas pemeliharaan untuk Managed Service untuk Apache Flink](maintenance.md).
Anda dapat memeriksa status aplikasi Anda menggunakan konsol, atau dengan menggunakan [DescribeApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DescribeApplication.html)tindakan.
+ **Job status** (Status tugas): Saat aplikasi Anda berada dalam status `RUNNING`, tugas Anda memiliki status yang menggambarkan fase eksekusi saat ini. Tugas dimulai dalam status `CREATED`, lalu meneruskan ke status `RUNNING` ketika sudah dimulai. Jika kondisi kesalahan terjadi, aplikasi Anda memasuki status berikut:
+ Untuk aplikasi yang menggunakan Apache Flink 1.11 dan yang lebih baru, aplikasi Anda memasuki status `RESTARTING`.
+ Untuk aplikasi yang menggunakan Apache Flink 1.8 dan sebelumnya, aplikasi Anda memasuki status `FAILING`.
Aplikasi selanjutnya meneruskan ke status `RESTARTING` atau `FAILED`, bergantung pada apakah tugas dapat dimulai ulang.
Anda dapat memeriksa status pekerjaan dengan memeriksa CloudWatch log aplikasi Anda untuk perubahan status.
## Jalankan beban kerja batch
Layanan Terkelola untuk Apache Flink mendukung menjalankan beban kerja batch Apache Flink. **Dalam pekerjaan batch, ketika pekerjaan Apache Flink mencapai status **SELESAI**, Layanan Terkelola untuk status aplikasi Apache Flink diatur ke READY.** Untuk informasi selengkapnya tentang status pekerjaan Flink, lihat [Pekerjaan dan](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/internals/job_scheduling/) Penjadwalan.
# Tinjau Layanan Terkelola untuk sumber daya aplikasi Apache Flink
Bagian ini menjelaskan sumber daya sistem yang digunakan aplikasi Anda. Memahami bagaimana Layanan Terkelola untuk penyediaan dan penggunaan sumber daya Apache Flink akan membantu Anda merancang, membuat, dan mempertahankan Layanan Terkelola yang berkinerja dan stabil untuk aplikasi Apache Flink.
## Layanan Terkelola untuk sumber daya aplikasi Apache Flink
Managed Service for Apache Flink adalah AWS layanan yang menciptakan lingkungan untuk hosting aplikasi Apache Flink Anda. Layanan Managed untuk layanan Apache Flink menyediakan sumber daya menggunakan unit yang disebut **Kinesis Processing** Unit (). KPUs
Satu KPU mewakili sumber daya sistem berikut:
+ Satu inti CPU
+ 4 GB memori, dengan satu GB adalah memori asli dan tiga GB adalah memori timbunan
+ 50 GB ruang disk
KPUs menjalankan aplikasi dalam unit eksekusi yang berbeda yang disebut **tugas** dan **subtask**. Anda bisa menganggap subtugas seperti utas.
Jumlah yang KPUs tersedia untuk aplikasi sama dengan `Parallelism` pengaturan aplikasi, dibagi dengan `ParallelismPerKPU` pengaturan aplikasi.
Untuk informasi selengkapnya tentang paralelisme aplikasi, lihat [Menerapkan penskalaan aplikasi](how-scaling.md).
## Sumber daya aplikasi Apache Flink
Lingkungan Apache Flink mengalokasikan sumber daya untuk aplikasi Anda menggunakan unit yang disebut **slot tugas**. Ketika Managed Service untuk Apache Flink mengalokasikan sumber daya untuk aplikasi Anda, ia menetapkan satu atau lebih slot tugas Apache Flink ke satu KPU. Jumlah slot yang ditetapkan ke satu KPU sama dengan pengaturan `ParallelismPerKPU` aplikasi Anda. Untuk informasi selengkapnya tentang slot tugas, lihat [Penjadwalan Pekerjaan di Dokumentasi](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/internals/job_scheduling/) Apache Flink.
### Paralelisme operator
Anda dapat mengatur jumlah maksimum subtugas yang dapat digunakan operator. Nilai ini disebut **Paralelisme Operator**. Secara default, paralelisme dari setiap operator dalam aplikasi Anda adalah sama dengan paralelisme aplikasi. Artinya, setiap operator dalam aplikasi Anda secara default dapat menggunakan semua subtugas yang tersedia dalam aplikasi jika diperlukan.
Anda dapat mengatur paralelisme operator dalam aplikasi Anda menggunakan metode `setParallelism`. Dengan menggunakan metode ini, Anda dapat mengontrol jumlah subtugas yang dapat digunakan setiap operator pada satu waktu.
Untuk informasi selengkapnya tentang operator, lihat [Operator](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/operators/overview/) di Dokumentasi Apache Flink.
### Rantai operator
Biasanya, setiap operator menggunakan subtugas terpisah untuk mengeksekusi, tetapi jika beberapa operator selalu mengeksekusi secara berurutan, runtime dapat menetapkannya ke tugas yang sama. Proses ini disebut **Rantai Operator**.
Beberapa operator berurutan dapat dirantai menjadi satu tugas jika semuanya beroperasi pada data yang sama. Berikut adalah beberapa kriteria yang diperlukan agar hal ini benar:
+ Operator melakukan penerusan sederhana 1 ke 1.
+ Operator semuanya memiliki paralelisme operator yang sama.
Ketika aplikasi Anda merantai operator menjadi satu subtugas, hal ini menghemat sumber daya sistem, karena layanan tidak perlu melakukan operasi jaringan dan mengalokasikan subtugas untuk setiap operator. Untuk menentukan apakah aplikasi Anda menggunakan rantai operator, lihat grafik pekerjaan di konsol Managed Service for Apache Flink. Setiap simpul dalam aplikasi mewakili satu atau beberapa operator. Grafik menunjukkan operator yang sudah dirantai sebagai satu simpul.
# Penagihan per detik di Managed Service untuk Apache Flink
Layanan Terkelola untuk Apache Flink sekarang ditagih dalam kenaikan satu detik. Ada biaya minimum sepuluh menit per aplikasi. Penagihan per detik berlaku untuk aplikasi yang baru diluncurkan atau sudah berjalan. Bagian ini menjelaskan bagaimana Layanan Terkelola untuk Apache Flink meteran dan menagih Anda untuk penggunaan Anda. Untuk mempelajari lebih lanjut tentang harga Layanan Terkelola untuk Apache Flink, lihat [Amazon Managed Service for Apache](https://aws.amazon.com/managed-service-apache-flink/pricing/) Flink Pricing.
## Cara kerjanya
Layanan Terkelola untuk Apache Flink menagih Anda untuk durasi dan jumlah **Unit Pemrosesan Kinesis (KPUs) yang ditagih secara bertahap satu detik di unit** yang didukung. Wilayah AWS KPU tunggal terdiri dari komputasi 1vCPU dan memori 4 GB. Anda dikenakan tarif per jam berdasarkan jumlah yang KPUs digunakan untuk menjalankan aplikasi Anda.
Misalnya, aplikasi yang berjalan selama 20 menit dan 10 detik akan dikenakan biaya selama 20 menit dan 10 detik, dikalikan dengan sumber daya yang digunakannya. Aplikasi yang berjalan selama 5 menit akan dikenakan biaya minimum sepuluh menit, dikalikan dengan sumber daya yang digunakannya.
Layanan Terkelola untuk Apache Flink menyatakan penggunaan dalam jam. Misalnya, 15 menit sesuai dengan 0,25 jam.
Untuk aplikasi Apache Flink, Anda dikenakan biaya satu KPU tambahan per aplikasi, yang digunakan untuk orkestrasi. Aplikasi juga dikenakan biaya untuk menjalankan penyimpanan dan cadangan yang tahan lama. Menjalankan penyimpanan aplikasi digunakan untuk kemampuan pemrosesan stateful di Managed Service untuk Apache Flink dan dikenakan biaya per. GB/month. Durable backups are optional and provide point-in-time recovery for applications, charged per GB/month
Dalam mode streaming, Managed Service untuk Apache Flink secara otomatis menskalakan jumlah yang KPUs dibutuhkan oleh aplikasi pemrosesan streaming Anda karena permintaan memori dan komputasi berfluktuasi. Anda dapat memilih untuk menyediakan aplikasi Anda dengan jumlah yang diperlukan KPUs.
## Wilayah AWS ketersediaan
**catatan**
Saat ini, tagihan per detik tidak tersedia di Wilayah berikut: AWS GovCloud (AS-Timur), (AS-Barat), Tiongkok AWS GovCloud (Beijing), dan Tiongkok (Ningxia).
Tagihan per detik tersedia sebagai berikut: Wilayah AWS
+ US East (N. Virginia) - us-east-1
+ US East (Ohio) - us-east-2
+ US West (N. California) - us-west-1
+ US West (Oregon) - us-west-2
+ Afrika (Cape Town) - af-south-1
+ Asia Pasifik (Hong Kong) - ap-east-1
+ Asia Pasifik (Hyderabad) - ap-south-1
+ Asia Pasifik (Jakarta) - ap-southeast-3
+ Asia Pasifik (Melbourne) - ap-southeast-4
+ Asia Pacific (Mumbai) - ap-south-1
+ Asia Pasifik (Osaka) - ap-northeast-3
+ Asia Pacific (Seoul) - ap-northeast-2
+ Asia Pacific (Singapore) - ap-southeast-1
+ Asia Pacific (Sydney) - ap-southeast-2
+ Asia Pacific (Tokyo) - ap-northeast-1
+ Canada (Central) - ca-central-1
+ Kanada Barat (Calgary) - ca-west-1
+ Europe (Frankfurt) - eu-central-1
+ Europe (Ireland) - eu-west-1
+ Europe (London) - eu-west-2
+ Eropa (Milan) - eu-south-1
+ Europe (Paris) - eu-west-3
+ Eropa (Spanyol) - eu-south-2
+ Europe (Stockholm) - eu-north-1
+ Eropa (Zürich) - eu-central-2
+ Israel (Tel Aviv) - il-central-1
+ Timur Tengah (Bahrain) - me-south-1
+ Timur Tengah (UEA) - me-central-1
+ South America (São Paulo) - sa-east-1
## Contoh harga
Anda dapat menemukan contoh harga di halaman harga Managed Service for Apache Flink. Untuk informasi selengkapnya, lihat [Amazon Managed Service untuk Harga Apache Flink](https://aws.amazon.com/managed-service-apache-flink/pricing/). Berikut ini adalah contoh lebih lanjut dengan ilustrasi Laporan Penggunaan Biaya untuk masing-masing ilustrasi.
### Beban kerja yang berjalan lama dan berat
Anda adalah layanan streaming Video besar dan Anda ingin membuat rekomendasi video real-time berdasarkan interaksi pengguna Anda. Anda menggunakan aplikasi Apache Flink di Managed Service untuk Apache Flink untuk terus mencerna peristiwa interaksi pengguna dari beberapa aliran data Kinesis dan untuk memproses peristiwa secara real time sebelum keluar ke sistem hilir. Peristiwa interaksi pengguna diubah menggunakan beberapa operator. Ini termasuk mempartisi data berdasarkan jenis peristiwa, memperkaya data dengan metadata tambahan, menyortir data berdasarkan stempel waktu, dan buffering data selama 5 menit sebelum pengiriman. Aplikasi ini memiliki banyak langkah transformasi yang intensif komputasi dan dapat diparalelkan. Aplikasi Flink Anda dikonfigurasi untuk berjalan dengan 20 KPUs untuk mengakomodasi beban kerja. Aplikasi Anda menggunakan cadangan aplikasi tahan lama 1 GB setiap hari. Biaya Managed Service bulanan untuk Apache Flink akan dihitung sebagai berikut:
**Biaya bulanan**
Harga di Wilayah AS Timur (Virginia N.) adalah \$10,11 per KPU-jam. Managed Service untuk Apache Flink mengalokasikan 50 GB penyimpanan aplikasi yang berjalan per KPU dengan biaya \$10,10 per GB/bulan.
+ Biaya KPU bulanan: 24 jam\$1 30 hari\$1 (20 KPUs \$1 1 tambahan KPU untuk aplikasi streaming) \$1 \$10,1/jam = \$11.584.00
+ Biaya penyimpanan aplikasi berjalan bulanan: 30 hari\$1 20 \$1 50 KPUs GB/KPUs \$1 \$10,10/GB-bulan = \$1100,00
+ Biaya penyimpanan aplikasi tahan lama bulanan: 30 hari\$1 1 GB \$1 0,023/GB-bulan = \$10,03
+ **Total biaya: \$11,584.00\$1\$1100 \$1 \$10,03 = \$11,684.03**
**Laporan penggunaan biaya untuk Managed Service untuk Apache Flink di konsol Billing and Cost Management untuk bulan tersebut**
Analisis Kinesis
+ USD 1,684.03 - AS Timur (Virginia Utara)
+ Amazon Kinesis Analytics CreateSnapshot
+ \$10,023 per GB-bulan cadangan aplikasi tahan lama
+ 1 GB-bulan - USD 0.03
+ Amazon Kinesis Analytics StartApplication
+ \$10,10 per GB-bulan penyimpanan aplikasi yang sedang berjalan
+ 1.000 GB-bulan - USD 100
+ \$10,11 per Unit Pemrosesan Kinesis - jam untuk aplikasi Apache Flink
+ 15,120 KPU-jam - USD 1,584
### Beban kerja batch yang berjalan selama \$1 15 menit setiap hari
Anda menggunakan aplikasi Apache Flink di Managed Service untuk Apache Flink untuk mengubah data log di Amazon Simple Storage Service (Amazon S3) dalam mode batch. Data log diubah menggunakan beberapa operator. Ini termasuk menerapkan skema ke peristiwa log yang berbeda, mempartisi data berdasarkan jenis peristiwa, dan menyortir data berdasarkan stempel waktu. Aplikasi ini memiliki banyak langkah transformasi, tetapi tidak ada yang intensif secara komputasi. Aplikasi ini menelan data pada 2.000 records/second selama 15 menit setiap hari dalam sebulan 30 hari. Anda tidak membuat cadangan aplikasi yang tahan lama. Biaya Managed Service bulanan untuk Apache Flink akan dihitung sebagai berikut:
**Biaya bulanan**
Harga di Wilayah AS Timur (Virginia N.) adalah \$10,11 per KPU-jam. Managed Service untuk Apache Flink mengalokasikan 50 GB penyimpanan aplikasi yang berjalan per KPU dengan biaya \$10,10 per GB/bulan.
+ Batch Workload: Selama 15 menit per hari, Managed Service untuk aplikasi Apache Flink memproses 2.000 records/second, which takes 2KPUs. 30 days/month \$1 15 minutes/day = 450 minutes/month
+ Biaya KPU bulanan: 450 minutes/month \$1 (2 KPUs \$1 1 tambahan KPU untuk aplikasi streaming) \$1 \$10,1/jam = \$12,48
+ Biaya penyimpanan aplikasi berjalan bulanan: 450 minutes/month \$1 2 \$1 50 KPUs GB/KPUs \$1 \$10,10/GB-bulan = \$10,11
+ **Total biaya: \$12,48\$10,11 = \$12,59**
**Laporan penggunaan biaya untuk Managed Service untuk Apache Flink di konsol Billing and Cost Management untuk bulan tersebut**
Analisis Kinesis
+ USD 2,59 - AS Timur (Virginia Utara)
+ Amazon Kinesis Analytics StartApplication
+ \$10,10 per GB-bulan untuk menjalankan pencadangan aplikasi
+ 1.042 GB-Bulan - USD 0.11
+ \$10,11 per Unit Pemrosesan Kinesis - jam untuk aplikasi Apache Flink
+ 22.5 KPU-jam - USD 2.48
### Aplikasi pengujian yang berhenti dan dimulai terus menerus pada jam yang sama, menarik beberapa biaya minimum
Anda adalah platform e-commerce besar yang memproses jutaan transaksi setiap hari. Anda ingin mengembangkan deteksi penipuan waktu nyata. Anda menggunakan aplikasi Apache Flink di Managed Service untuk Apache Flink untuk menyerap peristiwa transaksi dari Kinesis Data Streams dan memproses peristiwa secara real-time dengan berbagai langkah transformasi. Ini termasuk menggunakan jendela geser untuk menggabungkan peristiwa, mempartisi peristiwa berdasarkan jenis peristiwa, dan menerapkan aturan deteksi khusus untuk berbagai jenis peristiwa. Selama pengembangan, Anda memulai dan menghentikan aplikasi Anda beberapa kali untuk menguji dan men-debug perilaku. Ada kalanya aplikasi Anda hanya berjalan selama beberapa menit. Ada satu jam ketika Anda menguji aplikasi Anda dengan 4 KPUs dan aplikasi Anda tidak menggunakan cadangan aplikasi yang tahan lama:
+ Pada pukul 10:05, Anda memulai aplikasi Anda, yang berjalan selama 30 menit sebelum berhenti pada pukul 10:35.
+ Pada pukul 10:40, Anda memulai aplikasi lagi, yang berjalan selama 5 menit sebelum berhenti pada pukul 10:45.
+ Pada pukul 10:50, Anda memulai aplikasi lagi, yang berjalan selama 2 menit sebelum berhenti pada pukul 10:52.
Layanan Terkelola untuk Apache Flink mengenakan biaya minimal 10 menit penggunaan setiap kali aplikasi mulai berjalan. Layanan Terkelola bulanan untuk penggunaan Apache Flink untuk aplikasi Anda akan dihitung sebagai berikut:
+ Pertama kali aplikasi Anda dimulai dan berhenti: 30 menit penggunaan
+ Kedua kalinya aplikasi Anda dimulai dan berhenti: 10 menit penggunaan (aplikasi Anda berjalan selama 5 menit dibulatkan hingga pengisian minimum 10 menit)
+ Ketiga kalinya aplikasi Anda dimulai dan berhenti: 10 menit penggunaan (aplikasi Anda berjalan selama 2 menit, dibulatkan hingga pengisian minimum 10 menit)
Secara total, aplikasi Anda akan dikenakan biaya selama 50 menit penggunaan. Jika tidak ada waktu lain di bulan aplikasi Anda berjalan, biaya Layanan Terkelola bulanan untuk Apache Flink akan dihitung sebagai berikut:
**Biaya bulanan**
Harga di Wilayah AS Timur (Virginia N.) adalah \$10,11 per KPU-jam. Managed Service untuk Apache Flink mengalokasikan 50 GB penyimpanan aplikasi yang berjalan per KPU dengan biaya \$10,10 per GB/bulan.
+ Biaya bulanan KPU: 50 menit\$1 (4 KPUs \$1 1 tambahan KPU untuk aplikasi streaming) \$1 \$10,1/jam = \$10,46 (dibulatkan ke sen terdekat)
+ Biaya penyimpanan aplikasi berjalan bulanan: 50 menit \$1 4 \$1 50 KPUs GB/KPUs \$1 \$10,10/GB-bulan = \$10,03 (dibulatkan ke sen terdekat)
+ **Total biaya: \$10,46\$10,03 = \$10,49**
**Laporan penggunaan biaya untuk Managed Service untuk Apache Flink di konsol Billing and Cost Management untuk bulan tersebut**
Analisis Kinesis
+ USD 0,49 - AS Timur (Virginia Utara)
+ Amazon Kinesis Analytics StartApplication
+ \$10,10 per GB-bulan penyimpanan aplikasi yang sedang berjalan
+ 0.232 GB-Bulan - USD 0.03
+ \$10,11 per Unit Pemrosesan Kinesis - jam untuk aplikasi Apache Flink
+ 4.167 KPU-jam - USD 0.46
# Tinjau komponen DataStream API
Aplikasi Apache Flink Anda menggunakan [Apache Flink DataStream API](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/overview/) untuk mengubah data dalam aliran data.
Bagian ini menjelaskan berbagai komponen yang memindahkan, mengubah, dan melacak data:
+ [Gunakan konektor untuk memindahkan data dalam Layanan Terkelola untuk Apache Flink dengan API DataStream](how-connectors.md): Komponen ini memindahkan data antara aplikasi Anda dan sumber serta tujuan data eksternal.
+ [Mengubah data menggunakan operator di Managed Service untuk Apache Flink dengan API DataStream](how-operators.md): Komponen ini mengubah atau mengelompokkan elemen data dalam aplikasi Anda.
+ [Lacak peristiwa di Managed Service untuk Apache Flink menggunakan API DataStream](how-time.md): Topik ini menjelaskan bagaimana Managed Service for Apache Flink melacak peristiwa saat menggunakan API. DataStream
# Gunakan konektor untuk memindahkan data dalam Layanan Terkelola untuk Apache Flink dengan API DataStream
Di Amazon Managed Service for Apache Flink DataStream API, *konektor* adalah komponen perangkat lunak yang memindahkan data masuk dan keluar dari Layanan Terkelola untuk aplikasi Apache Flink. Konektor adalah integrasi fleksibel yang memungkinkan Anda membaca dari file dan direktori. Konektor terdiri dari modul lengkap untuk berinteraksi dengan layanan Amazon dan sistem pihak ketiga.
Tipe konektor termasuk berikut ini:
+ [Tambahkan sumber data streaming](how-sources.md): Berikan data ke aplikasi Anda dari Kinesis data stream, file, atau sumber data lainnya.
+ [Tulis data menggunakan sink](how-sinks.md): Kirim data dari aplikasi Anda ke aliran data Kinesis, aliran Firehose, atau tujuan data lainnya.
+ [Gunakan I/O Asinkron](how-async.md): Menyediakan akses asinkron ke sumber data (seperti basis data) untuk memperkaya peristiwa aliran.
## Konektor yang tersedia
Kerangka kerja Apache Flink berisi konektor untuk mengakses data dari berbagai sumber. Untuk informasi tentang konektor yang tersedia di kerangka kerja Apache Flink, lihat [Konektor](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/connectors/) di [Dokumentasi Apache Flink](https://nightlies.apache.org/flink/flink-docs-release-1.15/).
**Awas**
Jika Anda memiliki aplikasi yang berjalan di Flink 1.6, 1.8, 1.11 atau 1.13 dan ingin berjalan di Timur Tengah (UEA), Asia Pasifik (Hyderabad), Israel (Tel Aviv), Eropa (Zurich), Timur Tengah (UEA), Asia Pasifik (Melbourne) atau Asia Pasifik (Jakarta), Anda mungkin harus membangun kembali arsip aplikasi Anda dengan konektor yang diperbarui atau meningkatkan ke Flink 1.18.
Konektor Apache Flink disimpan di repositori open source mereka sendiri. Jika Anda memutakhirkan ke versi 1.18 atau yang lebih baru, Anda harus memperbarui dependensi Anda. Untuk mengakses repositori konektor Apache Flink AWS , lihat. [flink-connector-aws](https://github.com/apache/flink-connector-aws)
Sumber Kinesis sebelumnya `org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer` dihentikan dan mungkin dihapus dengan rilis Flink di masa depan. Gunakan [Sumber Kinesis](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/datastream/kinesis/#kinesis-streams-source) sebagai gantinya.
Tidak ada kompatibilitas status antara `FlinkKinesisConsumer` dan`KinesisStreamsSource`. Untuk detailnya, lihat [Memigrasi pekerjaan yang ada ke Sumber Aliran Kinesis baru](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/datastream/kinesis/#migrating-existing-jobs-to-new-kinesis-streams-source-from-kinesis-consumer) di dokumentasi Apache Flink.
Berikut ini adalah pedoman yang direkomendasikan:
**Peningkatan konektor**
| Versi Flink | Konektor yang digunakan | Resolusi |
| --- | --- | --- |
| 1.19, 1.20 | Sumber Kinesis | Saat memutakhirkan ke Managed Service untuk Apache Flink versi 1.19 dan 1.20, pastikan Anda menggunakan konektor sumber Kinesis Data Streams terbaru. Itu harus versi 5.0.0 atau yang lebih baru. Untuk informasi selengkapnya, lihat [Konektor Amazon Kinesis Data Streams](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kinesis/). |
| 1.19, 1.20 | Wastafel Kinesis | Saat memutakhirkan ke Layanan Terkelola untuk Apache Flink versi 1.19 dan 1.20, pastikan Anda menggunakan konektor sink Kinesis Data Streams terbaru. Itu harus versi 5.0.0 atau yang lebih baru. Untuk informasi lebih lanjut, lihat [Kinesis Streams](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/datastream/kinesis/#kinesis-streams-sink) Sink. |
| 1.19, 1.20 | Sumber Streams DynamoDB | Saat memutakhirkan ke Managed Service untuk Apache Flink versi 1.19 dan 1.20, pastikan Anda menggunakan konektor sumber DynamoDB Streams terbaru. Itu harus versi 5.0.0 atau yang lebih baru. Untuk informasi selengkapnya, lihat Konektor [Amazon DynamoDB](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/dynamodb/). |
| 1.19, 1.20 | DynamoDB Wastafel | Saat memutakhirkan ke Managed Service untuk Apache Flink versi 1.19 dan 1.20, pastikan Anda menggunakan konektor wastafel DynamoDB terbaru. Itu harus versi 5.0.0 atau yang lebih baru. Untuk informasi selengkapnya, lihat Konektor [Amazon DynamoDB](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/dynamodb/). |
| 1.19, 1.20 | Wastafel Amazon SQS | Saat memutakhirkan ke Managed Service untuk Apache Flink versi 1.19 dan 1.20, pastikan Anda menggunakan konektor wastafel Amazon SQS terbaru. Itu harus versi 5.0.0 atau yang lebih baru. Untuk informasi selengkapnya, lihat [Amazon SQS Sink.](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/sqs/) |
| 1.19, 1.20 | Layanan Dikelola Amazon untuk Prometheus Sink | Saat memutakhirkan ke Layanan Terkelola untuk Apache Flink versi 1.19 dan 1.20, pastikan Anda menggunakan Layanan Terkelola Amazon terbaru untuk konektor wastafel Prometheus. Itu harus versi 1.0.0 atau yang lebih baru. Untuk informasi lebih lanjut, lihat [Prometheus](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/prometheus/) Sink. |
# Tambahkan sumber data streaming ke Layanan Terkelola untuk Apache Flink
Apache Flink menyediakan konektor untuk membaca dari file, soket, koleksi, dan sumber kustom. Dalam kode aplikasi Anda, Anda menggunakan [sumber Apache Flink](https://nightlies.apache.org/flink/flink-docs-release-1.15/dev/datastream_api.html#data-sources) untuk menerima data dari aliran. Bagian ini menjelaskan sumber yang tersedia untuk layanan Amazon.
## Gunakan aliran data Kinesis
`KinesisStreamsSource`Menyediakan data streaming ke aplikasi Anda dari aliran data Amazon Kinesis.
### Buat `KinesisStreamsSource`
Contoh kode berikut mendemonstrasikan pembuatan `KinesisStreamsSource`:
```
// Configure the KinesisStreamsSource
Configuration sourceConfig = new Configuration();
sourceConfig.set(KinesisSourceConfigOptions.STREAM_INITIAL_POSITION, KinesisSourceConfigOptions.InitialPosition.TRIM_HORIZON); // This is optional, by default connector will read from LATEST
// Create a new KinesisStreamsSource to read from specified Kinesis Stream.
KinesisStreamsSource kdsSource =
KinesisStreamsSource.builder()
.setStreamArn("arn:aws:kinesis:us-east-1:123456789012:stream/test-stream")
.setSourceConfig(sourceConfig)
.setDeserializationSchema(new SimpleStringSchema())
.setKinesisShardAssigner(ShardAssignerFactory.uniformShardAssigner()) // This is optional, by default uniformShardAssigner will be used.
.build();
```
Untuk informasi selengkapnya tentang penggunaan`KinesisStreamsSource`, lihat Konektor [Amazon Kinesis Data](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kinesis/) Streams dalam [dokumentasi Apache Flink dan KinesisConnectors ](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/java/KinesisConnectors) contoh publik kami di Github.
### Buat `KinesisStreamsSource` yang menggunakan konsumen EFO
`KinesisStreamsSource`Sekarang mendukung [Enhanced Fan-Out (EFO](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/datastream/kinesis/)).
Jika konsumen Kinesis menggunakan EFO, layanan Kinesis Data Streams memberikan bandwidth khusus miliknya sendiri, bukan meminta konsumen berbagi bandwidth aliran tetap dengan konsumen lain yang membaca dari aliran.
Untuk informasi selengkapnya tentang penggunaan EFO dengan konsumen Kinesis, [lihat FLIP-128: Enhanced Fan Out](https://cwiki.apache.org/confluence/display/FLINK/FLIP-128%3A+Enhanced+Fan+Out+for+AWS+Kinesis+Consumers) untuk Konsumen Kinesis. AWS
Anda mengaktifkan konsumen EFO dengan mengatur parameter berikut pada konsumen Kinesis:
+ **READER\$1TYPE:** Setel parameter ini ke **EFO** agar aplikasi Anda menggunakan konsumen EFO untuk mengakses data Kinesis Data Stream.
+ **EFO\$1CONSUMER\$1NAME:** Atur parameter ini ke nilai string yang unik di antara konsumen aliran ini. Menggunakan kembali nama konsumen di Kinesis Data Stream yang sama akan menyebabkan konsumen sebelumnya yang menggunakan nama tersebut dihentikan.
Untuk mengonfigurasi `KinesisStreamsSource` untuk menggunakan EFO, tambahkan parameter berikut ke konsumen:
```
sourceConfig.set(KinesisSourceConfigOptions.READER_TYPE, KinesisSourceConfigOptions.ReaderType.EFO);
sourceConfig.set(KinesisSourceConfigOptions.EFO_CONSUMER_NAME, "my-flink-efo-consumer");
```
Untuk contoh aplikasi Managed Service for Apache Flink yang menggunakan konsumen EFO, lihat contoh Konektor [Kinesis publik](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/java/KinesisConnectors) kami di Github.
## Gunakan Amazon MSK
Sumber `KafkaSource` menyediakan data streaming ke aplikasi Anda dari topik Amazon MSK.
### Buat `KafkaSource`
Contoh kode berikut mendemonstrasikan pembuatan `KafkaSource`:
```
KafkaSource source = KafkaSource.builder()
.setBootstrapServers(brokers)
.setTopics("input-topic")
.setGroupId("my-group")
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
```
Untuk informasi selengkapnya tentang cara menggunakan `KafkaSource`, lihat [Replikasi MSK](earlier.md#example-msk).
# Menulis data menggunakan sink di Managed Service untuk Apache Flink
Dalam kode aplikasi Anda, Anda dapat menggunakan konektor [sink Apache Flink](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/datastream/overview/) untuk menulis ke sistem eksternal, termasuk AWS layanan, seperti Kinesis Data Streams dan DynamoDB.
Apache Flink juga menyediakan sink untuk file dan soket, dan Anda dapat menerapkan sink kustom. Di antara beberapa wastafel yang didukung, berikut ini sering digunakan:
## Gunakan aliran data Kinesis
Apache Flink memberikan informasi tentang [Konektor Kinesis Data Streams](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/datastream/kinesis/) di dokumentasi Apache Flink.
Untuk contoh aplikasi yang menggunakan Kinesis data stream untuk input dan output, lihat [Tutorial: Mulai menggunakan DataStream API di Managed Service untuk Apache Flink](getting-started.md).
## Gunakan Apache Kafka dan Amazon Managed Streaming untuk Apache Kafka (MSK)
[Konektor Apache Flink Kafka](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/datastream/kafka/#kafka-sink) memberikan dukungan ekstensif untuk menerbitkan data ke Apache Kafka dan Amazon MSK, termasuk jaminan persis sekali. Untuk mempelajari cara menulis ke Kafka, lihat [contoh Konektor Kafka dalam dokumentasi](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/java/KafkaConnectors) Apache Flink.
## Gunakan Amazon S3
Anda dapat menggunakan Apache Flink `StreamingFileSink` untuk menulis objek ke bucket Amazon S3.
Untuk contoh tentang cara menulis objek ke S3, lihat [Contoh: Menulis ke ember Amazon S3](earlier.md#examples-s3).
## Gunakan Firehose
`FlinkKinesisFirehoseProducer`[Ini adalah wastafel Apache Flink yang andal dan dapat diskalakan untuk menyimpan output aplikasi menggunakan layanan Firehose.](https://docs.aws.amazon.com/firehose/latest/dev/) Bagian ini menjelaskan cara menyiapkan proyek Maven untuk membuat dan menggunakan `FlinkKinesisFirehoseProducer`.
**Topics**
+ [Buat `FlinkKinesisFirehoseProducer`](#sinks-firehose-create)
+ [Contoh Kode `FlinkKinesisFirehoseProducer`](#sinks-firehose-sample)
### Buat `FlinkKinesisFirehoseProducer`
Contoh kode berikut mendemonstrasikan pembuatan `FlinkKinesisFirehoseProducer`:
```
Properties outputProperties = new Properties();
outputProperties.setProperty(ConsumerConfigConstants.AWS_REGION, region);
FlinkKinesisFirehoseProducer sink = new FlinkKinesisFirehoseProducer<>(outputStreamName, new SimpleStringSchema(), outputProperties);
```
### Contoh Kode `FlinkKinesisFirehoseProducer`
Contoh kode berikut menunjukkan cara membuat dan mengkonfigurasi `FlinkKinesisFirehoseProducer` dan mengirim data dari aliran data Apache Flink ke layanan Firehose.
```
package com.amazonaws.services.kinesisanalytics;
import com.amazonaws.services.kinesisanalytics.flink.connectors.config.ProducerConfigConstants;
import com.amazonaws.services.kinesisanalytics.flink.connectors.producer.FlinkKinesisFirehoseProducer;
import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer;
import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer;
import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants;
import java.io.IOException;
import java.util.Map;
import java.util.Properties;
public class StreamingJob {
private static final String region = "us-east-1";
private static final String inputStreamName = "ExampleInputStream";
private static final String outputStreamName = "ExampleOutputStream";
private static DataStream createSourceFromStaticConfig(StreamExecutionEnvironment env) {
Properties inputProperties = new Properties();
inputProperties.setProperty(ConsumerConfigConstants.AWS_REGION, region);
inputProperties.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST");
return env.addSource(new FlinkKinesisConsumer<>(inputStreamName, new SimpleStringSchema(), inputProperties));
}
private static DataStream createSourceFromApplicationProperties(StreamExecutionEnvironment env)
throws IOException {
Map applicationProperties = KinesisAnalyticsRuntime.getApplicationProperties();
return env.addSource(new FlinkKinesisConsumer<>(inputStreamName, new SimpleStringSchema(),
applicationProperties.get("ConsumerConfigProperties")));
}
private static FlinkKinesisFirehoseProducer createFirehoseSinkFromStaticConfig() {
/*
* com.amazonaws.services.kinesisanalytics.flink.connectors.config.
* ProducerConfigConstants
* lists of all of the properties that firehose sink can be configured with.
*/
Properties outputProperties = new Properties();
outputProperties.setProperty(ConsumerConfigConstants.AWS_REGION, region);
FlinkKinesisFirehoseProducer sink = new FlinkKinesisFirehoseProducer<>(outputStreamName,
new SimpleStringSchema(), outputProperties);
ProducerConfigConstants config = new ProducerConfigConstants();
return sink;
}
private static FlinkKinesisFirehoseProducer createFirehoseSinkFromApplicationProperties() throws IOException {
/*
* com.amazonaws.services.kinesisanalytics.flink.connectors.config.
* ProducerConfigConstants
* lists of all of the properties that firehose sink can be configured with.
*/
Map applicationProperties = KinesisAnalyticsRuntime.getApplicationProperties();
FlinkKinesisFirehoseProducer sink = new FlinkKinesisFirehoseProducer<>(outputStreamName,
new SimpleStringSchema(),
applicationProperties.get("ProducerConfigProperties"));
return sink;
}
public static void main(String[] args) throws Exception {
// set up the streaming execution environment
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
/*
* if you would like to use runtime configuration properties, uncomment the
* lines below
* DataStream input = createSourceFromApplicationProperties(env);
*/
DataStream input = createSourceFromStaticConfig(env);
// Kinesis Firehose sink
input.addSink(createFirehoseSinkFromStaticConfig());
// If you would like to use runtime configuration properties, uncomment the
// lines below
// input.addSink(createFirehoseSinkFromApplicationProperties());
env.execute("Flink Streaming Java API Skeleton");
}
}
```
Untuk tutorial lengkap tentang cara menggunakan wastafel Firehose, lihat. [Contoh: Menulis ke Firehose](earlier.md#get-started-exercise-fh)
# Gunakan Asynchronous I/O di Managed Service untuk Apache Flink
I/O Operator asinkron memperkaya data aliran menggunakan sumber data eksternal seperti database. Layanan Terkelola untuk Apache Flink memperkaya peristiwa streaming secara asinkron sehingga permintaan dapat dikumpulkan untuk efisiensi yang lebih besar.
Untuk informasi selengkapnya, lihat [Asynchronous I/O](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/operators/asyncio/) di Apache Flink Documentation.
# Mengubah data menggunakan operator di Managed Service untuk Apache Flink dengan API DataStream
*Untuk mengubah data masuk dalam Layanan Terkelola untuk Apache Flink, Anda menggunakan operator Apache Flink.* Operator Apache Flink mengubah satu atau beberapa aliran data menjadi aliran data baru. Aliran data baru berisi data yang dimodifikasi dari aliran data asli. Apache Flink menyediakan lebih dari 25 operator pemrosesan aliran yang dibangun sebelumnya. Untuk informasi selengkapnya, lihat [Operator](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/operators/overview/) di Dokumentasi Apache Flink.
**Topics**
+ [Gunakan operator transformasi](#how-operators-transform)
+ [Gunakan operator agregasi](#how-operators-agg)
## Gunakan operator transformasi
Berikut adalah contoh transformasi teks sederhana pada salah satu bidang aliran data JSON.
Kode ini membuat aliran data yang diubah. Aliran data baru memiliki data yang sama dengan aliran asli, dengan string "` Company`" yang ditambahkan ke isi bidang `TICKER`.
```
DataStream output = input.map(
new MapFunction() {
@Override
public ObjectNode map(ObjectNode value) throws Exception {
return value.put("TICKER", value.get("TICKER").asText() + " Company");
}
}
);
```
## Gunakan operator agregasi
Berikut adalah contoh operator agregasi. Kode membuat aliran data agregat. Operator membuat jendela tumbling 5 detik dan menampilkan jumlah dari nilai `PRICE` untuk catatan di jendela dengan nilai `TICKER` yang sama.
```
DataStream output = input.keyBy(node -> node.get("TICKER").asText())
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.reduce((node1, node2) -> {
double priceTotal = node1.get("PRICE").asDouble() + node2.get("PRICE").asDouble();
node1.replace("PRICE", JsonNodeFactory.instance.numberNode(priceTotal));
return node1;
});
```
Untuk contoh kode lainnya, lihat[Contoh untuk membuat dan bekerja dengan Managed Service untuk aplikasi Apache Flink](examples-collapsibles.md).
# Lacak peristiwa di Managed Service untuk Apache Flink menggunakan API DataStream
Layanan Terkelola untuk Apache Flink melacak peristiwa menggunakan stempel waktu berikut:
+ **Processing Time** (Waktu Pemrosesan): Mengacu pada waktu sistem mesin yang menjalankan operasi masing-masing.
+ **Event Time** (Waktu Peristiwa): Mengacu pada waktu setiap peristiwa individu terjadi pada perangkat produksinya.
+ **Waktu Tertelan:** Mengacu pada waktu peristiwa memasuki Layanan Terkelola untuk layanan Apache Flink.
Anda mengatur waktu yang digunakan oleh lingkungan streaming menggunakan`setStreamTimeCharacteristic`.
```
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
```
Untuk informasi selengkapnya tentang stempel waktu, lihat [Membuat Tanda Air di dokumentasi](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/event-time/generating_watermarks/) Apache Flink.
# Tinjau komponen API Tabel
Aplikasi Apache Flink Anda menggunakan [API Tabel Apache Flink](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/table/tableapi/) untuk berinteraksi dengan data dalam aliran menggunakan model relasional. Anda menggunakan API Tabel untuk mengakses data menggunakan sumber Tabel, lalu menggunakan fungsi Tabel untuk mengubah dan memfilter data tabel. Anda dapat mengubah dan memfilter data tabel menggunakan fungsi API atau perintah SQL.
Bagian ini berisi topik berikut:
+ [Konektor API tabel](how-table-connectors.md): Komponen ini memindahkan data antara aplikasi Anda dan sumber serta tujuan data eksternal.
+ [Atribut waktu API tabel](how-table-timeattributes.md): Topik ini menjelaskan bagaimana Managed Service for Apache Flink melacak peristiwa saat menggunakan Table API.
# Konektor API tabel
Dalam model pemrograman Apache Flink, konektor adalah komponen yang digunakan aplikasi Anda untuk membaca atau menulis data dari sumber eksternal, seperti layanan lain AWS .
Dengan API Tabel Apache Flink, Anda dapat menggunakan tipe konektor berikut:
+ [Sumber API tabel](#how-table-connectors-source): Anda menggunakan konektor sumber API Tabel untuk membuat tabel dalam `TableEnvironment` Anda menggunakan panggilan API atau kueri SQL.
+ [Tabel API tenggelam](#how-table-connectors-sink): Anda menggunakan perintah SQL untuk menulis data tabel ke sumber eksternal seperti topik Amazon MSK atau bucket Amazon S3.
## Sumber API tabel
Anda membuat sumber tabel dari aliran data. Kode berikut membuat tabel dari topik Amazon MSK:
```
//create the table
final FlinkKafkaConsumer consumer = new FlinkKafkaConsumer(kafkaTopic, new KafkaEventDeserializationSchema(), kafkaProperties);
consumer.setStartFromEarliest();
//Obtain stream
DataStream events = env.addSource(consumer);
Table table = streamTableEnvironment.fromDataStream(events);
```
Untuk informasi selengkapnya tentang sumber tabel, lihat [Konektor Tabel & SQL di Dokumentasi](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/overview/) Apache Flink.
## Tabel API tenggelam
Untuk menulis data tabel ke sink, Anda membuat sink di SQL, lalu jalankan sink berbasis SQL di objek `StreamTableEnvironment`.
Contoh kode berikut mendemonstrasikan cara menulis data tabel ke sink Amazon S3:
```
final String s3Sink = "CREATE TABLE sink_table (" +
"event_time TIMESTAMP," +
"ticker STRING," +
"price DOUBLE," +
"dt STRING," +
"hr STRING" +
")" +
" PARTITIONED BY (ticker,dt,hr)" +
" WITH" +
"(" +
" 'connector' = 'filesystem'," +
" 'path' = '" + s3Path + "'," +
" 'format' = 'json'" +
") ";
//send to s3
streamTableEnvironment.executeSql(s3Sink);
filteredTable.executeInsert("sink_table");
```
Anda dapat menggunakan `format` parameter untuk mengontrol format Managed Service untuk Apache Flink yang digunakan untuk menulis output ke wastafel. Untuk informasi tentang format, lihat [Konektor yang Didukung](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/overview/) di Dokumentasi Apache Flink.
## Sumber dan sink yang ditentukan pengguna
Anda dapat menggunakan konektor Apache Kafka yang ada untuk mengirim data ke dan dari layanan AWS lainnya, seperti Amazon MSK dan Amazon S3. Untuk berinteraksi dengan sumber data dan tujuan lainnya, Anda dapat menentukan sumber dan sink Anda sendiri. Untuk informasi selengkapnya, lihat [Sumber dan Tenggelam yang ditentukan pengguna](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/table/sourcessinks/) di Dokumentasi Apache Flink.
# Atribut waktu API tabel
Setiap catatan dalam aliran data memiliki beberapa stempel waktu yang menentukan kapan peristiwa yang terkait dengan catatan terjadi:
+ **Event Time** (Waktu Peristiwa): Stempel waktu yang ditetapkan pengguna yang menentukan kapan peristiwa yang dibuat catatan terjadi.
+ **Ingestion Time** (Waktu Penyerapan): Waktu ketika aplikasi Anda mengambil catatan dari aliran data.
+ **Processing Time** (Waktu pemrosesan): Waktu ketika aplikasi Anda memproses catatan.
Saat Apache Flink Table API membuat jendela berdasarkan waktu rekaman, Anda menentukan stempel waktu mana yang digunakan dengan menggunakan metode ini. `setStreamTimeCharacteristic`
Untuk informasi selengkapnya tentang penggunaan stempel waktu dengan API Tabel, lihat [Atribut Waktu](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/table/concepts/time_attributes/) dan [Pemrosesan Stream Tepat Waktu di Dokumentasi](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/concepts/time/) Apache Flink.
# Gunakan Python dengan Managed Service untuk Apache Flink
**catatan**
Jika Anda mengembangkan aplikasi Python Flink pada Mac baru dengan chip Apple Silicon, Anda mungkin mengalami beberapa masalah yang [diketahui dengan](https://issues.apache.org/jira/browse/FLINK-26981) dependensi Python 1,15. PyFlink Dalam hal ini kami sarankan menjalankan interpreter Python di Docker. Untuk step-by-step petunjuknya, lihat [PyFlink 1,15 pengembangan di Apple Silicon Mac](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/python/LocalDevelopmentOnAppleSilicon).
Apache Flink versi 2.2 mencakup dukungan untuk membuat aplikasi menggunakan Python versi 3.12; dukungan untuk Python versi 3.8 dihapus. Untuk informasi selengkapnya, lihat [Flink Python](https://nightlies.apache.org/flink/flink-docs-release-2.2/api/python/) Docs. Anda membuat Managed Service untuk aplikasi Apache Flink menggunakan Python dengan melakukan hal berikut:
+ Buat kode aplikasi Python Anda sebagai file teks dengan metode `main`.
+ Gabungkan file kode aplikasi Anda dan dependensi Python atau Java apa pun ke dalam file zip, dan unggah ke bucket Amazon S3.
+ Buat Layanan Terkelola untuk aplikasi Apache Flink Anda, tentukan lokasi kode Amazon S3, properti aplikasi, dan setelan aplikasi Anda.
Pada tingkat tinggi, API Tabel Python adalah wrapper di sekitar API Tabel Java. Untuk informasi tentang Python Table API, lihat Tabel API [Tutorial](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/python/table_api_tutorial/) di Apache Flink Documentation.
# Program Layanan Terkelola Anda untuk aplikasi Apache Flink Python
Anda kode Layanan Terkelola untuk Apache Flink untuk aplikasi Python menggunakan Apache Flink Python Table API. Mesin Apache Flink menerjemahkan pernyataan API Tabel Python (berjalan di VM Python) menjadi pernyataan API Tabel Java (berjalan di VM Java).
Anda menggunakan API Tabel Python dengan melakukan hal berikut:
+ Buat referensi ke`StreamTableEnvironment`.
+ Buat objek `table` dari data streaming sumber Anda dengan menjalankan query pada referensi `StreamTableEnvironment`.
+ Jalankan kueri di objek `table` untuk membuat tabel output.
+ Tulis tabel output Anda ke tujuan Anda menggunakan `StatementSet`.
Untuk mulai menggunakan Python Table API di Managed Service for Apache Flink, lihat. [Memulai Amazon Managed Service untuk Apache Flink untuk Python](gs-python.md)
## Membaca dan menulis data streaming
Untuk membaca dan menulis data streaming, Anda menjalankan kueri SQL pada lingkungan tabel.
### Membuat tabel
Contoh kode berikut menunjukkan fungsi yang ditetapkan pengguna yang membuat kueri SQL. Kueri SQL membuat tabel yang berinteraksi dengan aliran Kinesis:
```
def create_table(table_name, stream_name, region, stream_initpos):
return """ CREATE TABLE {0} (
`record_id` VARCHAR(64) NOT NULL,
`event_time` BIGINT NOT NULL,
`record_number` BIGINT NOT NULL,
`num_retries` BIGINT NOT NULL,
`verified` BOOLEAN NOT NULL
)
PARTITIONED BY (record_id)
WITH (
'connector' = 'kinesis',
'stream' = '{1}',
'aws.region' = '{2}',
'scan.stream.initpos' = '{3}',
'sink.partitioner-field-delimiter' = ';',
'sink.producer.collection-max-count' = '100',
'format' = 'json',
'json.timestamp-format.standard' = 'ISO-8601'
) """.format(table_name, stream_name, region, stream_initpos)
```
### Baca data streaming
Contoh kode berikut menunjukkan cara menggunakan kueri SQL `CreateTable` sebelumnya di referensi lingkungan tabel untuk membaca data:
```
table_env.execute_sql(create_table(input_table, input_stream, input_region, stream_initpos))
```
### Tulis data streaming
Contoh kode berikut menunjukkan cara menggunakan kueri SQL dari contoh `CreateTable` untuk membuat referensi tabel output, dan cara menggunakan `StatementSet` untuk berinteraksi dengan tabel untuk menulis data ke aliran Kinesis tujuan:
```
table_result = table_env.execute_sql("INSERT INTO {0} SELECT * FROM {1}"
.format(output_table_name, input_table_name))
```
## Baca properti runtime
Anda dapat menggunakan properti runtime untuk mengonfigurasi aplikasi Anda tanpa mengubah kode aplikasi Anda.
Anda menentukan properti aplikasi untuk aplikasi Anda dengan cara yang sama seperti dengan Managed Service untuk Apache Flink untuk aplikasi Java. Anda dapat menentukan properti runtime dengan cara berikut:
+ Menggunakan [CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html)tindakan.
+ Menggunakan [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)tindakan.
+ Mengonfigurasi aplikasi Anda menggunakan konsol.
Anda mengambil properti aplikasi dalam kode dengan membaca file json yang disebut `application_properties.json` bahwa runtime Layanan Terkelola untuk Apache Flink dibuat.
Contoh kode berikut menunjukkan properti aplikasi membaca dari file `application_properties.json`:
```
file_path = '/etc/flink/application_properties.json'
if os.path.isfile(file_path):
with open(file_path, 'r') as file:
contents = file.read()
properties = json.loads(contents)
```
Contoh kode fungsi yang ditetapkan pengguna berikut menunjukkan membaca grup properti dari objek properti aplikasi: mengambil:
```
def property_map(properties, property_group_id):
for prop in props:
if prop["PropertyGroupId"] == property_group_id:
return prop["PropertyMap"]
```
Contoh kode berikut menunjukkan membaca properti yang disebut INPUT\$1STREAM\$1KEY dari grup properti yang dikembalikan contoh sebelumnya:
```
input_stream = input_property_map[INPUT_STREAM_KEY]
```
## Buat paket kode aplikasi Anda
Setelah Anda membuat aplikasi Python, Anda menggabungkan file kode Anda dan dependensi ke dalam file zip.
File zip Anda harus berisi script python dengan metode `main`, dan secara opsional dapat berisi berikut ini:
+ File kode Python tambahan
+ Kode Java yang ditetapkan pengguna dalam file JAR
+ Pustaka Java dalam file JAR
**catatan**
File zip aplikasi Anda harus berisi semua dependensi untuk aplikasi Anda. Anda tidak dapat merujuk pustaka dari sumber lainnya untuk aplikasi Anda.
# Buat Layanan Terkelola Anda untuk aplikasi Apache Flink Python
## Tentukan file kode Anda
Setelah Anda telah membuat paket kode aplikasi, Anda mengunggahnya ke bucket Amazon S3. Anda kemudian membuat aplikasi Anda menggunakan konsol atau [CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html)tindakan.
Ketika Anda membuat aplikasi Anda menggunakan [CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html)tindakan, Anda menentukan file kode dan arsip dalam file zip Anda menggunakan grup properti aplikasi khusus yang disebut`kinesis.analytics.flink.run.options`. Anda dapat menentukan file tipe berikut:
+ **python**: File teks yang berisi metode utama Python.
+ **jarfile**: File JAR Java yang berisi fungsi yang ditetapkan pengguna Java.
+ **pyFiles**: File sumber daya Python yang berisi sumber daya yang akan digunakan oleh aplikasi.
+ **pyArchives**: File zip yang berisi file sumber daya untuk aplikasi.
Untuk informasi selengkapnya tentang jenis file kode Apache Flink Python, lihat [Antarmuka Baris Perintah di Dokumentasi Apache](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/cli/) Flink.
**catatan**
Layanan Terkelola untuk Apache Flink tidak mendukung`pyModule`,`pyExecutable`, atau jenis `pyRequirements` file. Semua kode, persyaratan, dan dependensi harus dalam file zip Anda. Anda tidak dapat menentukan dependensi yang akan diinstal menggunakan pip.
Cuplikan json contoh berikut menunjukkan cara menentukan lokasi file dalam file zip aplikasi Anda:
```
"ApplicationConfiguration": {
"EnvironmentProperties": {
"PropertyGroups": [
{
"PropertyGroupId": "kinesis.analytics.flink.run.options",
"PropertyMap": {
"python": "MyApplication/main.py",
"jarfile": "MyApplication/lib/myJarFile.jar",
"pyFiles": "MyApplication/lib/myDependentFile.py",
"pyArchives": "MyApplication/lib/myArchive.zip"
}
},
```
# Pantau Layanan Terkelola Anda untuk aplikasi Apache Flink Python
Anda menggunakan CloudWatch log aplikasi Anda untuk memantau Layanan Terkelola Anda untuk aplikasi Apache Flink Python.
Layanan Terkelola untuk Apache Flink mencatat pesan berikut untuk aplikasi Python:
+ Pesan yang ditulis ke konsol menggunakan `print()` di metode `main` aplikasi.
+ Pesan yang dikirim dalam fungsi yang ditetapkan pengguna menggunakan paket `logging`. Contoh kode berikut menunjukkan menulis ke log aplikasi dari fungsi yang ditetapkan pengguna:
```
import logging
@udf(input_types=[DataTypes.BIGINT()], result_type=DataTypes.BIGINT())
def doNothingUdf(i):
logging.info("Got {} in the doNothingUdf".format(str(i)))
return i
```
+ Pesan kesalahan yang dilemparkan oleh aplikasi.
Jika aplikasi melemparkan pengecualian di fungsi `main`, pengecualian akan muncul di log aplikasi Anda.
Contoh berikut menunjukkan entri log untuk pengecualian yang dilemparkan dari kode Python:
```
2021-03-15 16:21:20.000 --------------------------- Python Process Started --------------------------
2021-03-15 16:21:21.000 Traceback (most recent call last):
2021-03-15 16:21:21.000 " File ""/tmp/flink-web-6118109b-1cd2-439c-9dcd-218874197fa9/flink-web-upload/4390b233-75cb-4205-a532-441a2de83db3_code/PythonKinesisSink/PythonUdfUndeclared.py"", line 101, in "
2021-03-15 16:21:21.000 main()
2021-03-15 16:21:21.000 " File ""/tmp/flink-web-6118109b-1cd2-439c-9dcd-218874197fa9/flink-web-upload/4390b233-75cb-4205-a532-441a2de83db3_code/PythonKinesisSink/PythonUdfUndeclared.py"", line 54, in main"
2021-03-15 16:21:21.000 " table_env.register_function(""doNothingUdf"", doNothingUdf)"
2021-03-15 16:21:21.000 NameError: name 'doNothingUdf' is not defined
2021-03-15 16:21:21.000 --------------------------- Python Process Exited ---------------------------
2021-03-15 16:21:21.000 Run python process failed
2021-03-15 16:21:21.000 Error occurred when trying to start the job
```
**catatan**
Karena masalah performa, sebaiknya hanya gunakan pesan log kustom selama pengembangan aplikasi.
## Log kueri dengan CloudWatch Wawasan
Kueri CloudWatch Insights berikut mencari log yang dibuat oleh entrypoint Python saat menjalankan fungsi utama aplikasi Anda:
```
fields @timestamp, message
| sort @timestamp asc
| filter logger like /PythonDriver/
| limit 1000
```
# Menggunakan properti runtime di Managed Service untuk Apache Flink
Anda dapat menggunakan *properti runtime* untuk mengonfigurasi aplikasi Anda tanpa mengompilasi ulang kode aplikasi Anda.
**Topics**
+ [Mengelola properti runtime menggunakan konsol](#how-properties-console)
+ [Mengelola properti runtime menggunakan CLI](#how-properties-cli)
+ [Mengakses properti runtime dalam Layanan Terkelola untuk aplikasi Apache Flink](#how-properties-access)
## Mengelola properti runtime menggunakan konsol
Anda dapat menambahkan, memperbarui, atau menghapus properti runtime dari Layanan Terkelola untuk aplikasi Apache Flink Anda menggunakan. Konsol Manajemen AWS
**catatan**
Jika Anda menggunakan versi Apache Flink yang didukung sebelumnya dan ingin meningkatkan aplikasi yang ada ke Apache Flink 1.19.1, Anda dapat melakukannya menggunakan upgrade versi Apache Flink di tempat. Dengan peningkatan versi di tempat, Anda mempertahankan ketertelusuran aplikasi terhadap satu ARN di seluruh versi Apache Flink, termasuk snapshot, log, metrik, tag, konfigurasi Flink, dan banyak lagi. Anda dapat menggunakan fitur ini di `RUNNING` dan `READY` negara bagian. Untuk informasi selengkapnya, lihat [Gunakan upgrade versi di tempat untuk Apache Flink](how-in-place-version-upgrades.md).
**Perbarui Properti Runtime untuk Layanan Terkelola untuk aplikasi Apache Flink**
1. Masuk ke Konsol Manajemen AWS, dan buka konsol Amazon MSF di /flink. https://console.aws.amazon.com
1. Pilih Layanan Terkelola Anda untuk aplikasi Apache Flink. Pilih **Application details** (Detail aplikasi).
1. Di halaman untuk aplikasi Anda, pilih **Configure** (Konfigurasikan).
1. Perluas bagian **Properties** (Properti).
1. Gunakan kontrol di bagian **Properti** untuk menentukan grup properti dengan pasangan nilai kunci. Gunakan kontrol ini untuk menambah, memperbarui, atau menghapus grup properti dan properti runtime.
1. Pilih **Perbarui**.
## Mengelola properti runtime menggunakan CLI
Anda dapat menambahkan, memperbarui, atau menghapus properti runtime menggunakan [AWS CLI](https://docs.aws.amazon.com/cli).
Bagian ini mencakup permintaan contoh tindakan API untuk mengonfigurasi properti runtime aplikasi. Untuk informasi tentang cara menggunakan file JSON untuk input tindakan API, lihat [Layanan Terkelola untuk kode contoh API Apache Flink](api-examples.md).
**catatan**
Ganti ID akun sampel (*`012345678901`*) dalam contoh berikut dengan ID akun Anda.
### Tambahkan properti runtime saat membuat aplikasi
Permintaan contoh berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html) menambahkan dua grup properti runtime (`ProducerConfigProperties` dan `ConsumerConfigProperties`) saat Anda membuat aplikasi:
```
{
"ApplicationName": "MyApplication",
"ApplicationDescription": "my java test app",
"RuntimeEnvironment": "FLINK-1_19",
"ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
"ApplicationConfiguration": {
"ApplicationCodeConfiguration": {
"CodeContent": {
"S3ContentLocation": {
"BucketARN": "arn:aws:s3:::ka-app-code-username",
"FileKey": "java-getting-started-1.0.jar"
}
},
"CodeContentType": "ZIPFILE"
},
"EnvironmentProperties": {
"PropertyGroups": [
{
"PropertyGroupId": "ProducerConfigProperties",
"PropertyMap" : {
"flink.stream.initpos" : "LATEST",
"aws.region" : "us-west-2",
"AggregationEnabled" : "false"
}
},
{
"PropertyGroupId": "ConsumerConfigProperties",
"PropertyMap" : {
"aws.region" : "us-west-2"
}
}
]
}
}
}
```
### Menambahkan dan memperbarui properti runtime dalam aplikasi yang ada
Permintaan contoh berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) menambahkan atau memperbarui properti runtime untuk aplikasi yang ada:
```
{
"ApplicationName": "MyApplication",
"CurrentApplicationVersionId": 2,
"ApplicationConfigurationUpdate": {
"EnvironmentPropertyUpdates": {
"PropertyGroups": [
{
"PropertyGroupId": "ProducerConfigProperties",
"PropertyMap" : {
"flink.stream.initpos" : "LATEST",
"aws.region" : "us-west-2",
"AggregationEnabled" : "false"
}
},
{
"PropertyGroupId": "ConsumerConfigProperties",
"PropertyMap" : {
"aws.region" : "us-west-2"
}
}
]
}
}
}
```
**catatan**
Jika Anda menggunakan kunci yang tidak memiliki properti runtime yang sesuai dalam grup properti, Managed Service for Apache Flink menambahkan pasangan kunci-nilai sebagai properti baru. Jika Anda menggunakan kunci untuk properti runtime yang ada di grup properti, Managed Service for Apache Flink akan memperbarui nilai properti.
### Hapus properti runtime
Permintaan contoh berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) menghapus semua properti runtime dan grup properti dari aplikasi yang ada:
```
{
"ApplicationName": "MyApplication",
"CurrentApplicationVersionId": 3,
"ApplicationConfigurationUpdate": {
"EnvironmentPropertyUpdates": {
"PropertyGroups": []
}
}
}
```
**penting**
Jika Anda menghilangkan grup properti yang ada atau kunci properti yang ada di grup properti, grup properti atau properti akan dihapus.
## Mengakses properti runtime dalam Layanan Terkelola untuk aplikasi Apache Flink
Anda mengambil properti runtime dalam kode aplikasi Java Anda menggunakan metode `KinesisAnalyticsRuntime.getApplicationProperties()` statis, yang mengembalikan objek `Map`.
Contoh kode Java berikut mengambil properti runtime untuk aplikasi Anda:
```
Map applicationProperties = KinesisAnalyticsRuntime.getApplicationProperties();
```
Anda mengambil grup properti (sebagai objek `Java.Util.Properties`) sebagai berikut:
```
Properties consumerProperties = applicationProperties.get("ConsumerConfigProperties");
```
Anda biasanya mengonfigurasi sumber atau sink Apache Flinkdengan meneruskan di objek `Properties` tanpa perlu mengambil properti individu. Contoh kode berikut menunjukkan cara membuat sumber Flink dengan meneruskan di objek `Properties` yang diambil dari properti runtime:
```
private static FlinkKinesisProducer createSinkFromApplicationProperties() throws IOException {
Map applicationProperties = KinesisAnalyticsRuntime.getApplicationProperties();
FlinkKinesisProducer sink = new FlinkKinesisProducer(new SimpleStringSchema(),
applicationProperties.get("ProducerConfigProperties"));
sink.setDefaultStream(outputStreamName);
sink.setDefaultPartition("0");
return sink;
}
```
Untuk contoh kode, lihat [Contoh untuk membuat dan bekerja dengan Managed Service untuk aplikasi Apache Flink](examples-collapsibles.md).
# Gunakan konektor Apache Flink dengan Managed Service untuk Apache Flink
Konektor Apache Flink adalah komponen perangkat lunak yang memindahkan data masuk dan keluar dari Amazon Managed Service untuk aplikasi Apache Flink. Konektor adalah integrasi fleksibel yang memungkinkan Anda membaca dari file dan direktori. Konektor terdiri dari modul lengkap untuk berinteraksi dengan layanan Amazon dan sistem pihak ketiga.
Tipe konektor termasuk berikut ini:
+ **Sumber:** Berikan data ke aplikasi Anda dari aliran data Kinesis, file, topik Apache Kafka, file, atau sumber data lainnya.
+ **Tenggelam:** Kirim data dari aplikasi Anda ke aliran data Kinesis, aliran Firehose, topik Apache Kafka, atau tujuan data lainnya.
+ **Asynchronous I/O:** Menyediakan akses asinkron ke sumber data seperti database untuk memperkaya aliran.
Konektor Apache Flink disimpan di repositori sumbernya sendiri. Versi dan artefak untuk konektor Apache Flink berubah tergantung pada versi Apache Flink yang Anda gunakan, dan apakah Anda menggunakan, Table DataStream, atau SQL API.
Amazon Managed Service untuk Apache Flink mendukung lebih dari 40 sumber dan konektor sink Apache Flink yang sudah dibuat sebelumnya. Tabel berikut memberikan ringkasan konektor paling populer dan versi terkaitnya. Anda juga dapat membuat wastafel khusus menggunakan kerangka Async-sink. Untuk informasi selengkapnya, lihat [The Generic Asynchronous Base Sink](https://flink.apache.org/2022/03/16/the-generic-asynchronous-base-sink/) dalam dokumentasi Apache Flink.
Untuk mengakses repositori konektor Apache Flink AWS , lihat. [flink-connector-aws](https://github.com/apache/flink-connector-aws)
## Konektor untuk Flink 2.2
Saat memutakhirkan ke Flink 2.2, Anda perlu memperbarui dependensi konektor Anda ke versi yang kompatibel dengan runtime Flink 2.x. Konektor Flink dilepaskan secara independen dari runtime Flink, dan belum semua konektor memiliki rilis yang kompatibel dengan Flink 2.x. Tabel berikut merangkum ketersediaan konektor yang umum digunakan di Amazon Managed Service untuk Apache Flink pada tulisan ini:
**Konektor untuk Flink 2.2**
| Konektor | Flink 2.0\$1 Versi | Catatan |
| --- | --- | --- |
| Apache Kafka | flink-connector-kafka 4.0.0-2.0 | Direkomendasikan untuk Flink 2.2 |
| Kinesis Data Streams (sumber) | flink-connector-aws-kinesis-aliran 6.0.0-2.0 | Direkomendasikan untuk Flink 2.2 |
| Kinesis Data Streams (wastafel) | flink-connector-aws-kinesis-aliran 6.0.0-2.0 | Direkomendasikan untuk Flink 2.2 |
| FileSystem (S3, HDFS) | Dibundel dengan Flink | Dibangun ke dalam distribusi Flink - selalu tersedia |
| JDBC | Belum dirilis untuk 2.x | Tidak ada rilis yang kompatibel dengan Flink 2.x |
| OpenSearch | Belum dirilis untuk 2.x | Tidak ada rilis yang kompatibel dengan Flink 2.x |
| Elasticsearch | Belum dirilis untuk 2.x | Pertimbangkan untuk bermigrasi ke konektor OpenSearch |
| Amazon Managed Service for Prometheus | Belum dirilis untuk 2.x | Tidak ada rilis yang kompatibel dengan Flink 2.x pada saat penulisan |
Jika aplikasi Anda bergantung pada konektor yang belum memiliki rilis Flink 2.2, Anda memiliki dua opsi: tunggu konektor merilis versi yang kompatibel, atau evaluasi apakah Anda dapat menggantinya dengan alternatif (misalnya, menggunakan katalog JDBC atau wastafel khusus).
**Masalah yang diketahui**
+ Aplikasi yang menggunakan jalur `KinesisStreamsSource` with EFO (Enhanced Fan-Out/ SubscribeToShard) yang diperkenalkan di konektor v5.0.0 dan v6.0.0 mungkin gagal saat aliran Kinesis mengalami resharding. Ini adalah masalah yang diketahui di masyarakat. Untuk informasi lebih lanjut, lihat [FLINK-37648](https://issues.apache.org/jira/browse/FLINK-37648).
+ Aplikasi yang menggunakan jalur `KinesisStreamsSource` with EFO (Enhanced Fan-Out/ SubscribeToShard) yang diperkenalkan di konektor v5.0.0 dan v6.0.0 bersama-sama dengan `KinesisStreamsSink` mungkin mengalami kebuntuan jika aplikasi Flink berada di bawah tekanan balik, menghasilkan penghentian total pemrosesan data dalam satu atau lebih. TaskManagers Operasi penghentian paksa dan operasi aplikasi mulai diperlukan untuk memulihkan aplikasi. Ini adalah sub-kasus dari masalah yang diketahui di komunitas: [FLINK-34071](https://issues.apache.org/jira/browse/FLINK-34071).
## Konektor untuk versi Flink yang lebih lama
**Konektor untuk versi Flink yang lebih lama**
| Konektor | Flink versi 1.15 | Flink versi 1.18 | Flink versi 1.19 | Flink versi 1.20 |
| --- | --- | --- | --- | --- |
| Aliran Data Kinesis - Sumber - DataStream dan API Tabel | flink-connector-kinesis, 1.15.4 | flink-connector-kinesis, 4.3.0-1.18 | flink-connector-kinesis, 5.0.0-1.19 | flink-connector-kinesis, 5.0.0-1.20 |
| Aliran Data Kinesis - Sink - DataStream dan API Tabel | flink-connector-aws-kinesis-aliran, 1.15.4 | flink-connector-aws-kinesis-aliran, 4.3.0-1.18 | flink-connector-aws-kinesis-aliran, 5.0.0-1.19 | flink-connector-aws-kinesis-aliran, 5.0.0-1.20 |
| Kinesis Data Source/Sink Streams - - SQL | flink-sql-connector-kinesis, 1.15.4 | flink-sql-connector-kinesis, 4.3.0-1.18 | flink-sql-connector-kinesis, 5.0.0-1.19 | flink-sql-connector-kinesis-aliran, 5.0.0-1.20 |
| Kafka - DataStream dan Tabel API | flink-connector-kafka, 1.15.4 | flink-connector-kafka, 3.2.0-1.18 | flink-connector-kafka, 3.3.0-1.19 | flink-connector-kafka, 3.3.0-1.20 |
| Kafka - SQL | flink-sql-connector-kafka, 1.15.4 | flink-sql-connector-kafka, 3.2.0-1.18 | flink-sql-connector-kafka, 3.3.0-1.19 | flink-sql-connector-kafka, 3.3.0-1.20 |
| Firehose - DataStream dan Tabel API | flink-connector-aws-kinesis-selang api, 1.15.4 | flink-connector-aws-firehose, 4.3.0-1.18 | flink-connector-aws-firehose, 5.0.0-1.19 | flink-connector-aws-firehose, 5.0.0-1.20 |
| Firehose - SQL | flink-sql-connector-aws-kinesis-firehose, 1.15.4 | flink-sql-connector-aws-selang api, 4.3.0-1.18 | flink-sql-connector-aws-firehose, 5.0.0-1.19 | flink-sql-connector-aws-firehose, 5.0.0-1.20 |
| DynamoDB - dan Tabel API DataStream | flink-connector-dynamodb, 3.0.0-1.15 | flink-connector-dynamodb, 4.3.0-1.18 | flink-connector-dynamodb, 5.0.0-1.19 | flink-connector-dynamodb, 5.0.0-1.20 |
| DynamoDB - SQL | flink-sql-connector-dynamodb, 3.0.0-1.15 | flink-sql-connector-dynamodb, 4.3.0-1.18 | flink-sql-connector-dynamodb, 5.0.0-1.19 | flink-sql-connector-dynamodb, 5.0.0-1.20 |
| OpenSearch - DataStream dan Tabel API | - | flink-connector-opensearch, 1.2.0-1.18 | flink-connector-opensearch, 1.2.0-1.19 | flink-connector-opensearch, 1.2.0-1.19 |
| OpenSearch - SQL | - | flink-sql-connector-opensearch, 1.2.0-1.18 | flink-sql-connector-opensearch, 1.2.0-1.19 | flink-sql-connector-opensearch, 1.2.0-1.19 |
| Layanan Dikelola Amazon untuk Prometheus DataStream | - | flink-sql-connector-opensearch, 1.2.0-1.18 | flink-connector-prometheus, 1.0.0-1.19 | flink-connector-prometheus, 1.0.0-1.20 |
| Amazon SQS DataStream dan Tabel API | - | flink-sql-connector-opensearch, 1.2.0-1.18 | flink-connector-sqs, 5.0.0-1.19 | flink-connector-sqs, 5.0.0-1.20 |
Untuk mempelajari lebih lanjut tentang konektor di Amazon Managed Service untuk Apache Flink, lihat:
+ [DataStream Konektor API](https://docs.aws.amazon.com/managed-flink/latest/java/how-connectors.html)
+ [Konektor API tabel](https://docs.aws.amazon.com/managed-flink/latest/java/how-table-connectors.html)
### Masalah yang diketahui
Ada masalah open source Apache Flink yang diketahui dengan konektor Apache Kafka di Apache Flink 1.15. Masalah ini diselesaikan di versi Apache Flink yang lebih baru.
Lihat informasi yang lebih lengkap di [Masalah yang diketahui](flink-1-15-2.md#flink-1-15-known-issues).
# Menerapkan toleransi kesalahan dalam Layanan Terkelola untuk Apache Flink
Checkpointing adalah metode yang digunakan untuk menerapkan toleransi kesalahan di Amazon Managed Service untuk Apache Flink. *Pos pemeriksaan* adalah up-to-date cadangan dari aplikasi yang sedang berjalan yang digunakan untuk memulihkan segera dari gangguan atau kegagalan aplikasi yang tidak terduga.
Untuk detail tentang checkpointing di aplikasi Apache Flink, lihat [Checkpoints](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/ops/state/checkpoints/) di Apache Flink Documentation.
*Snapshot* adalah cadangan status aplikasi yang dibuat dan dikelola secara manual. Snapshot memungkinkan Anda memulihkan aplikasi Anda ke status sebelumnya dengan memanggil [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html). Untuk informasi selengkapnya, lihat [Kelola cadangan aplikasi menggunakan snapshot](how-snapshots.md).
Jika checkpointing diaktifkan untuk aplikasi Anda, layanan menyediakan toleransi kesalahan dengan membuat dan memuat cadangan data aplikasi jika terjadi mulai ulang aplikasi tak terduga. Mulai ulang aplikasi tak terduga ini dapat disebabkan oleh mulai ulang tugas tak terduga, kegagalan instans, dll. Ini memberi aplikasi semantik yang sama seperti eksekusi bebas kegagalan selama mulai ulang ini.
Jika snapshot diaktifkan untuk aplikasi, dan dikonfigurasi menggunakan aplikasi [ApplicationRestoreConfiguration](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ApplicationRestoreConfiguration.html), maka layanan menyediakan semantik pemrosesan tepat sekali selama pembaruan aplikasi, atau selama penskalaan atau pemeliharaan terkait layanan.
## Konfigurasikan checkpointing di Managed Service untuk Apache Flink
Anda dapat mengonfigurasi perilaku checkpointing aplikasi Anda. Anda dapat menentukan apakah ini mempertahankan status checkpointing, seberapa sering status untuk titik pemeriksaan disimpan, dan interval minimum antara akhir dari satu operasi titik pemeriksaan dan awal dari operasi lainnya.
Anda mengonfigurasi pengaturan berikut menggunakan operasi API [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html) atau [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html):
+ `CheckpointingEnabled` — Menunjukkan apakah checkpointing diaktifkan dalam aplikasi.
+ `CheckpointInterval` — Berisi waktu dalam milidetik di antara operasi titik pemeriksaan (persistensi).
+ `ConfigurationType` — Atur nilai ini ke `DEFAULT` untuk menggunakan perilaku checkpointing default. Atur nilai ini ke `CUSTOM` untuk mengonfigurasi nilai lainnya.
**catatan**
Perilaku titik pemeriksaan default adalah sebagai berikut:
**CheckpointingEnabled:** benar
**CheckpointInterval:** 60000
**MinPauseBetweenCheckpoints:** 5000
Jika **ConfigurationType**diatur ke`DEFAULT`, nilai sebelumnya akan digunakan, bahkan jika mereka diatur ke nilai lain menggunakan baik menggunakan AWS Command Line Interface, atau dengan menetapkan nilai-nilai dalam kode aplikasi.
**catatan**
Untuk Flink 1.15 dan seterusnya, Layanan Terkelola untuk Apache Flink akan digunakan `stop-with-savepoint` selama Pembuatan Snapshot Otomatis, yaitu pembaruan aplikasi, penskalaan, atau penghentian.
+ `MinPauseBetweenCheckpoints` — Waktu minimum dalam milidetik antara akhir dari satu operasi titik pemeriksaan dan awal dari operasi lainnya. Mengatur nilai ini mencegah aplikasi dari melakukan checkpointing terus-menerus ketika operasi titik pemeriksaan memakan waktu lebih lama dari `CheckpointInterval`.
## Tinjau contoh API pos pemeriksaan
Bagian ini mencakup contoh permintaan tindakan API untuk mengonfigurasi checkpointing untuk aplikasi. Untuk informasi tentang cara menggunakan file JSON untuk input tindakan API, lihat [Layanan Terkelola untuk kode contoh API Apache Flink](api-examples.md).
### Konfigurasikan checkpointing untuk aplikasi baru
Contoh permintaan untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html) berikut mengonfigurasi checkpointing saat Anda membuat aplikasi:
```
{
"ApplicationName": "MyApplication",
"RuntimeEnvironment":"FLINK-1_19",
"ServiceExecutionRole":"arn:aws:iam::123456789123:role/myrole",
"ApplicationConfiguration": {
"ApplicationCodeConfiguration":{
"CodeContent":{
"S3ContentLocation":{
"BucketARN":"arn:aws:s3:::amzn-s3-demo-bucket",
"FileKey":"myflink.jar",
"ObjectVersion":"AbCdEfGhIjKlMnOpQrStUvWxYz12345"
}
},
"FlinkApplicationConfiguration": {
"CheckpointConfiguration": {
"CheckpointingEnabled": "true",
"CheckpointInterval": 20000,
"ConfigurationType": "CUSTOM",
"MinPauseBetweenCheckpoints": 10000
}
}
}
```
### Nonaktifkan checkpointing untuk aplikasi baru
Contoh permintaan untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html) berikut menonaktifkan checkpointing saat Anda membuat aplikasi:
```
{
"ApplicationName": "MyApplication",
"RuntimeEnvironment":"FLINK-1_19",
"ServiceExecutionRole":"arn:aws:iam::123456789123:role/myrole",
"ApplicationConfiguration": {
"ApplicationCodeConfiguration":{
"CodeContent":{
"S3ContentLocation":{
"BucketARN":"arn:aws:s3:::amzn-s3-demo-bucket",
"FileKey":"myflink.jar",
"ObjectVersion":"AbCdEfGhIjKlMnOpQrStUvWxYz12345"
}
},
"FlinkApplicationConfiguration": {
"CheckpointConfiguration": {
"CheckpointingEnabled": "false"
}
}
}
```
### Konfigurasikan checkpointing untuk aplikasi yang sudah ada
Contoh permintaan untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) berikut mengonfigurasi checkpointing untuk aplikasi yang ada:
```
{
"ApplicationName": "MyApplication",
"ApplicationConfigurationUpdate": {
"FlinkApplicationConfigurationUpdate": {
"CheckpointConfigurationUpdate": {
"CheckpointingEnabledUpdate": true,
"CheckpointIntervalUpdate": 20000,
"ConfigurationTypeUpdate": "CUSTOM",
"MinPauseBetweenCheckpointsUpdate": 10000
}
}
}
}
```
### Nonaktifkan checkpointing untuk aplikasi yang sudah ada
Contoh permintaan untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) berikut menonaktifkan checkpointing untuk aplikasi yang ada:
```
{
"ApplicationName": "MyApplication",
"ApplicationConfigurationUpdate": {
"FlinkApplicationConfigurationUpdate": {
"CheckpointConfigurationUpdate": {
"CheckpointingEnabledUpdate": false,
"CheckpointIntervalUpdate": 20000,
"ConfigurationTypeUpdate": "CUSTOM",
"MinPauseBetweenCheckpointsUpdate": 10000
}
}
}
}
```
# Kelola cadangan aplikasi menggunakan snapshot
*Snapshot* *adalah Managed Service untuk implementasi Apache Flink dari Apache Flink Savepoint.* Snapshot adalah cadangan status aplikasi yang dipicu, dibuat, dan dikelola pengguna atau layanan. [Untuk informasi tentang Apache Flink Savepoints, lihat Savepoints di Dokumentasi Apache Flink.](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/ops/state/savepoints/) Menggunakan snapshot, Anda dapat me-restart aplikasi dari snapshot tertentu dari status aplikasi.
**catatan**
Sebaiknya aplikasi Anda membuat snapshot beberapa kali sehari untuk memulai ulang dengan benar menggunakan data status yang benar. Frekuensi yang benar untuk snapshot Anda bergantung pada logika bisnis aplikasi Anda. Mengambil snapshot yang sering memungkinkan Anda memulihkan data yang lebih baru, tetapi meningkatkan biaya dan membutuhkan lebih banyak sumber daya sistem.
Di Managed Service for Apache Flink, Anda mengelola snapshot menggunakan tindakan API berikut:
+ [https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplicationSnapshot.html](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplicationSnapshot.html)
+ [https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DeleteApplicationSnapshot.html](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DeleteApplicationSnapshot.html)
+ [https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DescribeApplicationSnapshot.html](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DescribeApplicationSnapshot.html)
+ [https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ListApplicationSnapshots.html](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ListApplicationSnapshots.html)
Untuk batas per aplikasi pada jumlah snapshot, lihat [Layanan Terkelola untuk kuota notebook Apache Flink dan Studio](limits.md). Jika aplikasi Anda mencapai batas pada snapshot, lalu secara manual membuat snapshot gagal dengan `LimitExceededException`.
Layanan Terkelola untuk Apache Flink tidak pernah menghapus snapshot. Anda harus secara manual menghapus snapshot menggunakan tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DeleteApplicationSnapshot.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DeleteApplicationSnapshot.html).
Untuk memuat snapshot status aplikasi tersimpan saat memulai aplikasi, gunakan parameter [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ApplicationRestoreConfiguration.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ApplicationRestoreConfiguration.html) dari [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html) atau tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html).
**Topics**
+ [Kelola pembuatan snapshot otomatis](#how-fault-snapshot-update)
+ [Pulihkan dari snapshot yang berisi data status yang tidak kompatibel](#how-fault-snapshot-restore)
+ [Tinjau contoh API snapshot](#how-fault-snapshot-examples)
## Kelola pembuatan snapshot otomatis
Jika `SnapshotsEnabled` diatur ke `true` dalam untuk aplikasi, Managed Service [ ApplicationSnapshotConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ApplicationSnapshotConfiguration.html)for Apache Flink secara otomatis membuat dan menggunakan snapshot saat aplikasi diperbarui, diskalakan, atau dihentikan untuk menyediakan semantik pemrosesan yang tepat sekali.
**catatan**
Mengatur `ApplicationSnapshotConfiguration::SnapshotsEnabled` ke `false` akan menyebabkan kehilangan data selama pembaruan aplikasi.
**catatan**
Layanan Terkelola untuk Apache Flink memicu savepoint perantara selama pembuatan snapshot. Untuk Flink versi 1.15 atau lebih besar, savepoint menengah tidak lagi melakukan efek samping apa pun. Lihat [Memicu savepoint](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
Snapshot yang dibuat secara otomatis memiliki kualitas berikut:
+ Snapshot dikelola oleh layanan, tetapi Anda dapat melihat snapshot menggunakan tindakan. [ ListApplicationSnapshots](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ListApplicationSnapshots.html) Snapshot yang dibuat secara otomatis menghitung batas snapshot Anda.
+ Jika aplikasi Anda melebihi batas snapshot, snapshot yang dibuat secara manual akan gagal, tetapi Layanan Terkelola untuk layanan Apache Flink akan tetap berhasil membuat snapshot saat aplikasi diperbarui, diskalakan, atau dihentikan. Anda harus menghapus snapshot secara manual menggunakan [ DeleteApplicationSnapshot](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DeleteApplicationSnapshot.html)tindakan sebelum membuat lebih banyak snapshot secara manual.
## Pulihkan dari snapshot yang berisi data status yang tidak kompatibel
Karena snapshot berisi informasi tentang operator, memulihkan data status dari snapshot untuk operator yang telah berubah sejak versi aplikasi sebelumnya mungkin memiliki hasil yang tak terduga. Aplikasi akan gagal jika mencoba memulihkan data status dari snapshot yang tidak sesuai dengan operator saat ini. Aplikasi yang gagal akan terhenti di status `STOPPING` atau `UPDATING`.
Untuk memungkinkan aplikasi memulihkan dari snapshot yang berisi data status yang tidak kompatibel, atur `AllowNonRestoredState` parameter [FlinkRunConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_FlinkRunConfiguration.html)untuk `true` menggunakan tindakan. [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)
Anda akan melihat perilaku berikut ketika aplikasi dipulihkan dari snapshot usang:
+ **Operator ditambahkan:** Jika operator baru ditambahkan, titik simpan tidak memiliki data status untuk operator baru. Tidak ada kesalahan yang akan terjadi, dan tidak perlu untuk mengatur `AllowNonRestoredState`.
+ **Operator dihapus:** Jika operator yang ada dihapus, titik simpan memiliki data status untuk operator yang hilang. Kesalahan akan terjadi kecuali `AllowNonRestoredState` diatur ke `true`.
+ **Operator dimodifikasi:** Jika perubahan yang kompatibel dibuat, seperti mengubah tipe parameter ke tipe yang kompatibel, aplikasi dapat memulihkan dari snapshot usang. Untuk informasi selengkapnya tentang memulihkan dari snapshot, lihat [Savepoints](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/ops/state/savepoints/) di Dokumentasi Apache Flink. Aplikasi yang menggunakan Apache Flink versi 1.8 atau yang lebih baru mungkin dapat dipulihkan dari snapshot dengan skema yang berbeda. Aplikasi yang menggunakan Apache Flink versi 1.6 tidak dapat dipulihkan. Untuk two-phase-commit sink, sebaiknya gunakan snapshot sistem (SWs) alih-alih snapshot () buatan pengguna. CreateApplicationSnapshot
Untuk Flink, Layanan Terkelola untuk Apache Flink memicu savepoint perantara selama pembuatan snapshot. Untuk Flink 1.15 dan seterusnya, savepoint menengah tidak lagi melakukan efek samping apa pun. Lihat [Memicu Savepoint](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
Jika Anda perlu melanjutkan aplikasi yang tidak kompatibel dengan data savepoint yang ada, sebaiknya Anda melewatkan pemulihan dari snapshot dengan menyetel `ApplicationRestoreType` parameter tindakan ke. [StartApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StartApplication.html)`SKIP_RESTORE_FROM_SNAPSHOT`
Untuk informasi selengkapnya tentang cara Apache Flink menangai data status yang tidak kompatibel, lihat [Evolusi Skema Status](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/fault-tolerance/serialization/schema_evolution/) di *Dokumentasi Apache Flink*.
## Tinjau contoh API snapshot
Bagian ini mencakup permintaan contoh tindakan API untuk menggunakan snapshot dengan aplikasi. Untuk informasi tentang cara menggunakan file JSON untuk input tindakan API, lihat [Layanan Terkelola untuk kode contoh API Apache Flink](api-examples.md).
### Aktifkan snapshot untuk aplikasi
Contoh permintaan untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) berikut mengaktifkan snapshot untuk aplikasi:
```
{
"ApplicationName": "MyApplication",
"CurrentApplicationVersionId": 1,
"ApplicationConfigurationUpdate": {
"ApplicationSnapshotConfigurationUpdate": {
"SnapshotsEnabledUpdate": "true"
}
}
}
```
### Buat snapshot
Contoh permintaan untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplicationSnapshot.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplicationSnapshot.html) berikut membuat snapshot dari status aplikasi saat ini:
```
{
"ApplicationName": "MyApplication",
"SnapshotName": "MyCustomSnapshot"
}
```
### Buat daftar snapshot untuk aplikasi
Contoh permintaan untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplicationSnapshots.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplicationSnapshots.html) berikut mencantumkan 50 snapshot pertama untuk status aplikasi saat ini:
```
{
"ApplicationName": "MyApplication",
"Limit": 50
}
```
### Rincian daftar untuk snapshot aplikasi
Contoh permintaan berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplicationSnapshot.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplicationSnapshot.html) mencantumkan detail untuk snapshot aplikasi tertentu:
```
{
"ApplicationName": "MyApplication",
"SnapshotName": "MyCustomSnapshot"
}
```
### Menghapus snapshot
Contoh permintaan berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DeleteApplicationSnapshot.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DeleteApplicationSnapshot.html) menghapus snapshot yang disimpan sebelumnya. Anda bisa mendapatkan nilai `SnapshotCreationTimestamp` menggunakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplicationSnapshots.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplicationSnapshots.html) atau [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DeleteApplicationSnapshot.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DeleteApplicationSnapshot.html):
```
{
"ApplicationName": "MyApplication",
"SnapshotName": "MyCustomSnapshot",
"SnapshotCreationTimestamp": 12345678901.0,
}
```
### Mulai ulang aplikasi menggunakan snapshot bernama
Contoh permintaan berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html) memulai aplikasi menggunakan status yang disimpan dari snapshot tertentu:
```
{
"ApplicationName": "MyApplication",
"RunConfiguration": {
"ApplicationRestoreConfiguration": {
"ApplicationRestoreType": "RESTORE_FROM_CUSTOM_SNAPSHOT",
"SnapshotName": "MyCustomSnapshot"
}
}
}
```
### Mulai ulang aplikasi menggunakan snapshot terbaru
Contoh permintaan berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html) memulai aplikasi menggunakan snapshot terbaru:
```
{
"ApplicationName": "MyApplication",
"RunConfiguration": {
"ApplicationRestoreConfiguration": {
"ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
}
}
}
```
### Mulai ulang aplikasi tanpa snapshot
Contoh permintaan berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StartApplication.html) memulai aplikasi tanpa memuat status aplikasi, bahkan jika snapshot tersedia:
```
{
"ApplicationName": "MyApplication",
"RunConfiguration": {
"ApplicationRestoreConfiguration": {
"ApplicationRestoreType": "SKIP_RESTORE_FROM_SNAPSHOT"
}
}
}
```
# Gunakan upgrade versi di tempat untuk Apache Flink
Dengan upgrade versi in-place untuk Apache Flink, Anda mempertahankan ketertelusuran aplikasi terhadap satu ARN di seluruh versi Apache Flink. Ini termasuk snapshot, log, metrik, tag, konfigurasi Flink, peningkatan batas sumber daya, dan banyak lagi. VPCs
Anda dapat melakukan peningkatan versi di tempat untuk Apache Flink untuk meningkatkan aplikasi yang ada ke versi Flink baru di Amazon Managed Service untuk Apache Flink. Untuk melakukan tugas ini, Anda dapat menggunakan AWS CLI, AWS CloudFormation, AWS SDK, atau file. Konsol Manajemen AWS
**catatan**
Anda tidak dapat menggunakan upgrade versi di tempat untuk Apache Flink dengan Amazon Managed Service untuk Apache Flink Studio.
**Topics**
+ [Tingkatkan aplikasi menggunakan peningkatan versi di tempat untuk Apache Flink](upgrading-applications.md)
+ [Tingkatkan aplikasi Anda ke versi Apache Flink baru](upgrading-application-new-version.md)
+ [Kembalikan upgrade aplikasi](rollback.md)
+ [Praktik dan rekomendasi terbaik umum untuk peningkatan aplikasi](best-practices-recommendations.md)
+ [Tindakan pencegahan dan masalah yang diketahui dengan peningkatan aplikasi](precautions.md)
+ [Upgrade ke Flink 2.2: Panduan lengkap](flink-2-2-upgrade-guide.md)
+ [Panduan kompatibilitas status untuk peningkatan Flink 2.2](state-compatibility.md)
# Tingkatkan aplikasi menggunakan peningkatan versi di tempat untuk Apache Flink
Sebelum Anda mulai, kami sarankan Anda menonton video ini: [Upgrade Versi In-Place](https://www.youtube.com/watch?v=f1qGGdaP2XI).
Untuk melakukan upgrade versi di tempat untuk Apache Flink, Anda dapat menggunakan,, AWS SDK AWS CLI AWS CloudFormation, atau file. Konsol Manajemen AWS Anda dapat menggunakan fitur ini dengan aplikasi apa pun yang ada yang Anda gunakan dengan Managed Service for Apache Flink di negara bagian `READY` atau`RUNNING`. Ini menggunakan UpdateApplication API untuk menambahkan kemampuan untuk mengubah runtime Flink.
## Sebelum memutakhirkan: Perbarui aplikasi Apache Flink Anda
Saat Anda menulis aplikasi Flink Anda, Anda menggabungkannya dengan dependensinya ke dalam JAR aplikasi dan mengunggah JAR ke bucket Amazon S3 Anda. Dari sana, Amazon Managed Service untuk Apache Flink menjalankan pekerjaan di runtime Flink baru yang telah Anda pilih. Anda mungkin harus memperbarui aplikasi Anda untuk mencapai kompatibilitas dengan runtime Flink yang ingin Anda tingkatkan. Mungkin ada ketidakkonsistenan antara versi Flink yang menyebabkan peningkatan versi gagal. Paling umum, ini akan dengan konektor untuk sumber (masuknya) atau tujuan (sink, jalan keluar) dan dependensi Scala. Flink 1.15 dan versi yang lebih baru di Managed Service untuk Apache Flink adalah Scala-agnostik, dan JAR Anda harus berisi versi Scala yang Anda rencanakan untuk digunakan.
**Untuk memperbarui aplikasi Anda**
1. Baca saran dari komunitas Flink tentang peningkatan aplikasi dengan status. Lihat [Memutakhirkan Aplikasi dan Versi Flink](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/upgrading/).
1. Baca daftar mengetahui masalah dan batasan. Lihat [Tindakan pencegahan dan masalah yang diketahui dengan peningkatan aplikasi](precautions.md).
1. Perbarui dependensi Anda dan uji aplikasi Anda secara lokal. Dependensi ini biasanya adalah:
1. Runtime dan API Flink.
1. Konektor direkomendasikan untuk runtime Flink baru. Anda dapat menemukannya di [versi Rilis](https://docs.aws.amazon.com/managed-flink/latest/java/release-version-list.html) untuk runtime tertentu yang ingin Anda perbarui.
1. Scala - Apache Flink adalah Scala-agnostik dimulai dengan dan termasuk Flink 1.15. Anda harus menyertakan dependensi Scala yang ingin Anda gunakan dalam JAR aplikasi Anda.
1. Buat JAR aplikasi baru di zipfile dan unggah ke Amazon S3. Kami menyarankan Anda menggunakan nama yang berbeda dari JAR/ZipFile sebelumnya. Jika Anda perlu memutar kembali, Anda akan menggunakan informasi ini.
1. Jika Anda menjalankan aplikasi stateful, kami sangat menyarankan Anda mengambil snapshot dari aplikasi Anda saat ini. Ini memungkinkan Anda memutar kembali secara statis jika Anda mengalami masalah selama atau setelah peningkatan.
# Tingkatkan aplikasi Anda ke versi Apache Flink baru
Anda dapat meningkatkan aplikasi Flink Anda dengan menggunakan [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)tindakan.
Anda dapat memanggil `UpdateApplication` API dengan berbagai cara:
+ Gunakan alur kerja **Konfigurasi** yang ada di file. Konsol Manajemen AWS
+ Buka halaman aplikasi Anda di file Konsol Manajemen AWS.
+ Pilih **Konfigurasikan**
+ Pilih runtime baru dan snapshot yang ingin Anda mulai, juga dikenal sebagai konfigurasi pemulihan. Gunakan pengaturan terbaru sebagai konfigurasi pemulihan untuk memulai aplikasi dari snapshot terbaru. Arahkan ke aplikasi baru yang ditingkatkan JAR/zip di Amazon S3.
+ Gunakan tindakan AWS CLI [update-aplikasi](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesisanalyticsv2/update-application.html).
+ Gunakan CloudFormation (CFN).
+ Perbarui [RuntimeEnvironment](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-kinesisanalyticsv2-application.html#cfn-kinesisanalyticsv2-application-runtimeenvironment)bidang. Sebelumnya, CloudFormation menghapus aplikasi dan membuat yang baru, menyebabkan snapshot Anda dan riwayat aplikasi lainnya hilang. Sekarang CloudFormation perbarui RuntimeEnvironment tempat Anda dan tidak menghapus aplikasi Anda.
+ Gunakan AWS SDK.
+ Konsultasikan dokumentasi SDK untuk bahasa pemrograman pilihan Anda. Lihat [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html).
Anda dapat melakukan pemutakhiran saat aplikasi dalam `RUNNING` keadaan atau saat aplikasi dihentikan dalam `READY` status. Amazon Managed Service for Apache Flink memvalidasi untuk memverifikasi kompatibilitas antara versi runtime asli dan versi runtime target. Pemeriksaan kompatibilitas ini berjalan saat Anda melakukan [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)saat dalam `RUNNING` status atau berikutnya [StartApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StartApplication.html)jika Anda memutakhirkan saat dalam `READY` status.
## Tingkatkan aplikasi dalam `RUNNING` keadaan
Contoh berikut menunjukkan peningkatan aplikasi dalam `RUNNING` status bernama `UpgradeTest` Flink 1.18 di US East (Virginia N.) menggunakan AWS CLI dan memulai aplikasi yang ditingkatkan dari snapshot terbaru.
```
aws --region us-east-1 kinesisanalyticsv2 update-application \
--application-name UpgradeTest --runtime-environment-update "FLINK-1_18" \
--application-configuration-update '{"ApplicationCodeConfigurationUpdate": '\
'{"CodeContentUpdate": {"S3ContentLocationUpdate": '\
'{"FileKeyUpdate": "flink_1_18_app.jar"}}}}' \
--run-configuration-update '{"ApplicationRestoreConfiguration": '\
'{"ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"}}' \
--current-application-version-id ${current_application_version}
```
+ Jika Anda mengaktifkan snapshot layanan dan ingin melanjutkan aplikasi dari snapshot terbaru, Amazon Managed Service for Apache Flink memverifikasi bahwa runtime `RUNNING` aplikasi saat ini kompatibel dengan runtime target yang dipilih.
+ Jika Anda telah menetapkan snapshot untuk melanjutkan runtime target, Amazon Managed Service for Apache Flink memverifikasi bahwa runtime target kompatibel dengan snapshot yang ditentukan. Jika pemeriksaan kompatibilitas gagal, permintaan pembaruan Anda ditolak dan aplikasi Anda tetap tidak tersentuh dalam `RUNNING` status.
+ Jika Anda memilih untuk memulai aplikasi tanpa snapshot, Amazon Managed Service untuk Apache Flink tidak menjalankan pemeriksaan kompatibilitas apa pun.
+ Jika aplikasi Anda yang ditingkatkan gagal atau macet dalam `UPDATING` keadaan transitif, ikuti instruksi di [Kembalikan upgrade aplikasi](rollback.md) bagian untuk kembali ke keadaan sehat.
**Alur proses untuk menjalankan aplikasi status**
![\[Diagram berikut mewakili alur kerja yang disarankan untuk meng-upgrade aplikasi saat berjalan. Kami berasumsi bahwa aplikasi stateful dan Anda mengaktifkan snapshot. Untuk alur kerja ini, pada pembaruan, Anda memulihkan aplikasi dari snapshot terbaru yang secara otomatis diambil oleh Amazon Managed Service untuk Apache Flink sebelum memperbarui.\]](http://docs.aws.amazon.com/id_id/managed-flink/latest/java/images/in-place-update-while-running.png)
## Tingkatkan aplikasi dalam keadaan **READY**
Contoh berikut menunjukkan peningkatan aplikasi dalam `READY` status bernama `UpgradeTest` Flink 1.18 di US East (Virginia N.) menggunakan file. AWS CLI Tidak ada snapshot yang ditentukan untuk memulai aplikasi karena aplikasi tidak berjalan. Anda dapat menentukan snapshot saat mengeluarkan permintaan aplikasi mulai.
```
aws --region us-east-1 kinesisanalyticsv2 update-application \
--application-name UpgradeTest --runtime-environment-update "FLINK-1_18" \
--application-configuration-update '{"ApplicationCodeConfigurationUpdate": '\
'{"CodeContentUpdate": {"S3ContentLocationUpdate": '\
'{"FileKeyUpdate": "flink_1_18_app.jar"}}}}' \
--current-application-version-id ${current_application_version}
```
+ Anda dapat memperbarui runtime aplikasi Anda dalam `READY` status ke versi Flink apa pun. Amazon Managed Service untuk Apache Flink tidak menjalankan pemeriksaan apapun sampai Anda memulai aplikasi Anda.
+ Amazon Managed Service untuk Apache Flink hanya menjalankan pemeriksaan kompatibilitas terhadap snapshot yang Anda pilih untuk memulai aplikasi. Ini adalah pemeriksaan kompatibilitas dasar mengikuti [Tabel Kompatibilitas Flink](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/upgrading/#compatibility-table). Mereka hanya memeriksa versi Flink yang dengannya snapshot diambil dan versi Flink yang Anda targetkan. Jika runtime Flink dari snapshot yang dipilih tidak kompatibel dengan runtime baru aplikasi, permintaan mulai mungkin ditolak.
**Alur proses untuk aplikasi status siap**
![\[Diagram berikut mewakili alur kerja yang direkomendasikan untuk meng-upgrade aplikasi saat dalam keadaan siap. Kami berasumsi bahwa aplikasi stateful dan Anda mengaktifkan snapshot. Untuk alur kerja ini, pada pembaruan, Anda memulihkan aplikasi dari snapshot terbaru yang secara otomatis diambil oleh Amazon Managed Service untuk Apache Flink ketika aplikasi dihentikan.\]](http://docs.aws.amazon.com/id_id/managed-flink/latest/java/images/in-place-update-while-ready.png)
# Kembalikan upgrade aplikasi
Jika Anda memiliki masalah dengan aplikasi Anda atau menemukan inkonsistensi dalam kode aplikasi Anda antara versi Flink, Anda dapat memutar kembali menggunakan,, AWS SDK AWS CLI AWS CloudFormation, atau. Konsol Manajemen AWS Contoh berikut menunjukkan seperti apa rolling back dalam skenario kegagalan yang berbeda.
## Upgrade runtime berhasil, aplikasi dalam `RUNNING` keadaan, tetapi pekerjaan gagal dan terus dimulai ulang
Asumsikan Anda mencoba untuk meng-upgrade aplikasi stateful bernama `TestApplication` dari Flink 1.15 ke Flink 1.18 di US East (N. Virginia). Namun, aplikasi Flink 1.18 yang ditingkatkan gagal untuk memulai atau terus-menerus memulai ulang, meskipun aplikasi dalam keadaan. `RUNNING` Ini adalah skenario kegagalan yang umum. Untuk menghindari downtime lebih lanjut, kami sarankan Anda segera mengembalikan aplikasi Anda ke versi yang berjalan sebelumnya (Flink 1.15), dan mendiagnosis masalah nanti.
Untuk memutar kembali aplikasi ke versi berjalan sebelumnya, gunakan AWS CLI perintah [rollback-application](https://docs.aws.amazon.com/cli/latest/reference/kinesisanalyticsv2/rollback-application.html) atau tindakan API. [RollbackApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_RollbackApplication.html) Tindakan API ini mengembalikan perubahan yang Anda buat yang menghasilkan versi terbaru. Kemudian restart aplikasi Anda menggunakan snapshot sukses terbaru.
Kami sangat menyarankan Anda mengambil snapshot dengan aplikasi yang ada sebelum Anda mencoba untuk meningkatkan. Ini akan membantu menghindari kehilangan data atau harus memproses ulang data.
Dalam skenario kegagalan ini, tidak CloudFormation akan memutar kembali aplikasi untuk Anda. Anda harus memperbarui CloudFormation template untuk menunjuk ke runtime sebelumnya dan ke kode sebelumnya CloudFormation untuk memaksa memperbarui aplikasi. Jika tidak, CloudFormation asumsikan bahwa aplikasi Anda telah diperbarui saat transisi ke status. `RUNNING`
## Memutar kembali aplikasi yang macet `UPDATING`
Jika aplikasi Anda macet dalam `AUTOSCALING` status `UPDATING` atau setelah upaya upgrade, Amazon Managed Service untuk Apache Flink menawarkan AWS CLI perintah [rollback-applications](https://docs.aws.amazon.com/cli/latest/reference/kinesisanalyticsv2/rollback-application.html), atau tindakan [RollbackApplications](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_RollbackApplication.html)API yang dapat memutar kembali aplikasi ke versi sebelum macet atau status. `UPDATING` `AUTOSCALING` API ini mengembalikan perubahan yang Anda buat yang menyebabkan aplikasi macet `UPDATING` atau status `AUTOSCALING` transitif.
# Praktik dan rekomendasi terbaik umum untuk peningkatan aplikasi
+ Uji status baru job/runtime tanpa status pada lingkungan non-produksi sebelum mencoba peningkatan produksi.
+ Pertimbangkan untuk menguji peningkatan stateful dengan aplikasi non-produksi terlebih dahulu.
+ Pastikan grafik pekerjaan baru Anda memiliki status yang kompatibel dengan snapshot yang akan Anda gunakan untuk memulai aplikasi yang ditingkatkan.
+ Pastikan bahwa jenis yang disimpan dalam status operator tetap sama. Jika jenisnya telah berubah, Apache Flink tidak dapat memulihkan status operator.
+ Pastikan bahwa Operator yang IDs Anda atur menggunakan `uid` metode tetap sama. Apache Flink memiliki rekomendasi kuat untuk menetapkan unik IDs untuk operator. Untuk informasi selengkapnya, lihat [Menetapkan Operator IDs](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#assigning-operator-ids) di dokumentasi Apache Flink.
Jika Anda tidak menetapkan IDs ke operator Anda, Flink secara otomatis menghasilkannya. Dalam hal ini, mereka mungkin bergantung pada struktur program dan, jika diubah, dapat menyebabkan masalah kompatibilitas. Flink menggunakan Operator IDs untuk mencocokkan status dalam snapshot ke operator. Mengubah Operator IDs mengakibatkan aplikasi tidak dimulai, atau status yang disimpan dalam snapshot yang dijatuhkan, dan operator baru memulai tanpa status.
+ Jangan mengubah kunci yang digunakan untuk menyimpan status yang dikunci.
+ Jangan mengubah jenis input operator stateful seperti window atau join. Ini secara implisit mengubah jenis keadaan internal operator, menyebabkan ketidakcocokan status.
# Tindakan pencegahan dan masalah yang diketahui dengan peningkatan aplikasi
## Kafka Commit pada checkpointing gagal berulang kali setelah broker memulai ulang
Ada masalah Apache Flink open source yang diketahui dengan konektor Apache Kafka di Flink versi 1.15 yang disebabkan oleh bug Kafka Client open source kritis di Kafka Client 2.8.1. Untuk informasi selengkapnya, lihat [Komit Kafka pada pos pemeriksaan gagal berulang kali setelah broker memulai ulang](https://issues.apache.org/jira/browse/FLINK-28060) dan [KafkaConsumer tidak dapat memulihkan koneksi ke koordinator grup](https://issues.apache.org/jira/browse/KAFKA-13840) setelah pengecualian. commitOffsetAsync
Untuk menghindari masalah ini, kami sarankan Anda menggunakan Apache Flink 1.18 atau yang lebih baru di Amazon Managed Service untuk Apache Flink.
## Keterbatasan kompatibilitas negara yang diketahui
+ Jika Anda menggunakan Table API, Apache Flink tidak menjamin kompatibilitas status antara versi Flink. Untuk informasi selengkapnya, lihat [Peningkatan dan Evolusi Stateful](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/concepts/overview/#stateful-upgrades-and-evolution) dalam dokumentasi Apache Flink.
+ Status Flink 1.6 tidak kompatibel dengan Flink 1.18. API menolak permintaan Anda jika Anda mencoba memutakhirkan dari 1,6 ke 1,18 dan yang lebih baru dengan status. Anda dapat meningkatkan ke 1.8, 1.11, 1.13 dan 1.15 dan mengambil snapshot, dan kemudian meningkatkan ke 1.18 dan yang lebih baru. Untuk informasi selengkapnya, lihat [Upgrade Aplikasi dan Versi Flink di dokumentasi](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/upgrading/) Apache Flink.
## Masalah yang diketahui dengan Konektor Kinesis Flink
+ Jika Anda menggunakan Flink 1.11 atau lebih lama dan menggunakan `amazon-kinesis-connector-flink` konektor untuk dukungan Enhanced-fan-out (EFO), Anda harus mengambil langkah ekstra untuk upgrade stateful ke Flink 1.13 atau yang lebih baru. Ini karena perubahan nama paket konektor. Untuk informasi selengkapnya, lihat [amazon-kinesis-connector-flink](https://github.com/awslabs/amazon-kinesis-connector-flink).
`amazon-kinesis-connector-flink`Konektor untuk Flink 1.11 dan sebelumnya menggunakan kemasan`software.amazon.kinesis`, sedangkan konektor Kinesis untuk Flink 1.13 dan yang lebih baru menggunakan. `org.apache.flink.streaming.connectors.kinesis` Gunakan alat ini untuk mendukung migrasi Anda: [amazon-kinesis-connector-flink-state-migrator](https://github.com/awslabs/amazon-kinesis-connector-flink-state-migrator).
+ Jika Anda menggunakan Flink 1.13 atau lebih lama dengan `FlinkKinesisProducer` dan meningkatkan ke Flink 1.15 atau yang lebih baru, untuk peningkatan stateful Anda harus terus menggunakan `FlinkKinesisProducer` di Flink 1.15 atau yang lebih baru, bukan yang lebih baru. `KinesisStreamsSink` Namun, jika Anda sudah memiliki `uid` set khusus di wastafel Anda, Anda harus dapat beralih ke `KinesisStreamsSink` karena `FlinkKinesisProducer` tidak mempertahankan status. Flink akan memperlakukannya sebagai operator yang sama karena kustom `uid` diatur.
## Aplikasi Flink ditulis dalam Scala
+ Pada Flink 1.15, Apache Flink tidak menyertakan Scala dalam runtime. Anda harus menyertakan versi Scala yang ingin Anda gunakan dan dependensi Scala lainnya dalam kode Anda JAR/zip saat memutakhirkan ke Flink 1.15 atau yang lebih baru. Untuk informasi selengkapnya, lihat [Amazon Managed Service untuk Apache Flink untuk rilis Apache Flink 1.15.2](https://docs.aws.amazon.com/managed-flink/latest/java/flink-1-15-2.html).
+ Jika aplikasi Anda menggunakan Scala dan Anda memutakhirkannya dari Flink 1.11 atau sebelumnya (Scala 2.11) ke Flink 1.13 (Scala 2.12), pastikan kode Anda menggunakan Scala 2.12. Jika tidak, aplikasi Flink 1.13 Anda mungkin gagal menemukan kelas Scala 2.11 di runtime Flink 1.13.
## Hal-hal yang perlu dipertimbangkan saat menurunkan aplikasi Flink
+ Menurunkan aplikasi Flink dimungkinkan, tetapi terbatas pada kasus ketika aplikasi sebelumnya berjalan dengan versi Flink yang lebih lama. Untuk upgrade stateful Managed Service untuk Apache Flink akan memerlukan penggunaan snapshot yang diambil dengan versi pencocokan atau versi sebelumnya untuk downgrade
+ Jika Anda memperbarui runtime dari Flink 1.13 atau yang lebih baru ke Flink 1.11 atau yang lebih lama, dan jika aplikasi Anda menggunakan backend HashMap status, aplikasi Anda akan terus gagal.
# Upgrade ke Flink 2.2: Panduan lengkap
Panduan ini memberikan step-by-step petunjuk untuk meningkatkan Layanan Terkelola Amazon Anda untuk aplikasi Apache Flink dari Flink 1.x ke Flink 2.2. Ini adalah peningkatan versi utama dengan perubahan yang melanggar yang memerlukan perencanaan dan pengujian yang cermat.
**Upgrade versi utama adalah uni-directional**
Operasi Upgrade dapat memindahkan aplikasi Anda dari Flink 1.x ke 2.2 dengan pelestarian status, tetapi Anda tidak dapat bergerak kembali dari 2.2 ke 1.x dengan status 2.2. Jika aplikasi Anda menjadi tidak sehat setelah memutakhirkan, gunakan Rollback API untuk kembali ke versi 1.x dengan status 1.x asli Anda dari snapshot terbaru.
## Prasyarat
Sebelum memulai upgrade Anda:
+ Ulasan [Melanggar perubahan dan penghentian](flink-2-2.md#flink-2-2-breaking-changes)
+ Ulasan [Panduan kompatibilitas status untuk peningkatan Flink 2.2](state-compatibility.md)
+ Pastikan Anda memiliki lingkungan non-produksi untuk pengujian
+ Dokumentasikan konfigurasi dan dependensi aplikasi Anda saat ini
## Memahami jalur migrasi Anda
Pengalaman upgrade Anda bergantung pada kompatibilitas aplikasi Anda dengan Flink 2.2. Memahami jalur ini membantu Anda mempersiapkan dengan tepat dan menetapkan harapan yang realistis.
**Jalur 1: Status biner dan aplikasi yang kompatibel**
**Apa yang diharapkan:**
+ Memanggil operasi Upgrade
+ Selesaikan migrasi ke 2.2 dengan transisi status aplikasi: → → `RUNNING` `UPDATING` `RUNNING`
+ Pertahankan semua status aplikasi tanpa kehilangan data atau pemrosesan ulang
+ Pengalaman yang sama dengan migrasi versi minor
Terbaik untuk: Aplikasi atau aplikasi stateless menggunakan serialisasi yang kompatibel (Avro, skema Protobuf yang kompatibel, tanpa koleksi) POJOs
**Jalur 2: Ketidakcocokan biner**
**Apa yang diharapkan:**
+ Memanggil operasi Upgrade
+ Operasi gagal dan memunculkan ketidakcocokan biner melalui API Operasi dan log
+ Dengan auto-rollback diaktifkan: Aplikasi secara otomatis berputar kembali dalam beberapa menit tanpa intervensi Anda
+ Dengan auto-rollback dinonaktifkan: Aplikasi tetap dalam keadaan berjalan tanpa pemrosesan data; Anda secara manual memutar kembali ke versi yang lebih lama
+ Setelah biner diperbaiki, gunakan [UpdateApplication API](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html) untuk pengalaman yang mirip dengan Path 1
Terbaik untuk: Aplikasi yang menggunakan dihapus APIs yang terdeteksi selama startup pekerjaan Flink
**Jalur 3: Status aplikasi yang tidak kompatibel**
**Apa yang diharapkan:**
+ Memanggil operasi Upgrade
+ Migrasi tampaknya berhasil pada awalnya
+ Aplikasi memasuki loop restart dalam hitungan detik karena pemulihan status gagal
+ Mendeteksi kegagalan melalui CloudWatch Metrik yang menunjukkan restart terus menerus
+ Memanggil operasi Rollback secara manual
+ Kembali ke produksi dalam beberapa menit setelah memulai rollback
+ Tinjau [Migrasi negara](state-compatibility.md#state-compat-migration) untuk aplikasi Anda
Terbaik untuk: Aplikasi dengan ketidakcocokan serialisasi status (POJOs dengan koleksi, status serial KRYO tertentu)
**catatan**
Sangat disarankan untuk membuat replika aplikasi produksi Anda dan menguji setiap fase peningkatan berikut pada replika sebelum mengikuti langkah yang sama untuk aplikasi produksi Anda.
## Tahap 1: Persiapan
**Perbarui kode aplikasi**
Perbarui kode aplikasi Anda agar kompatibel dengan Flink 2.2:
+ **Perbarui dependensi Flink** ke versi 2.2.0 di atau `pom.xml` `build.gradle`
+ **Perbarui dependensi konektor ke versi** yang kompatibel dengan Flink 2.2 (lihat) [Ketersediaan konektor](flink-2-2.md#flink-2-2-connectors)
+ **Hapus penggunaan API yang tidak digunakan lagi**:
+ Ganti DataSet API dengan DataStream API atau Tabel API/SQL
+ Ganti `SourceFunction` `SinkFunction` lawasan/dengan FLIP-27 Source dan FLIP-143 Sink APIs
+ Ganti penggunaan Scala API dengan Java API
+ **Perbarui ke Java 17**
**Unggah kode aplikasi yang diperbarui**
+ Bangun JAR aplikasi Anda dengan dependensi Flink 2.2
+ Unggah ke Amazon S3 dengan **nama file yang berbeda** dari JAR Anda saat ini (misalnya,) `my-app-flink-2.2.jar`
+ Perhatikan bucket dan kunci S3 untuk digunakan pada langkah upgrade
## Tahap 2: Aktifkan rollback otomatis
Auto-rollback memungkinkan Amazon Managed Service untuk Apache Flink untuk secara otomatis kembali ke versi sebelumnya jika upgrade gagal.
**Periksa status rollback otomatis**
*Konsol Manajemen AWS:*
1. Arahkan ke aplikasi Anda
1. Pilih **Konfigurasi**
1. Di bawah **Pengaturan aplikasi**, verifikasi **Rollback sistem** diaktifkan
*AWS CLI:*
```
aws kinesisanalyticsv2 describe-application \
--application-name MyApplication \
--query 'ApplicationDetail.ApplicationConfigurationDescription.ApplicationSystemRollbackConfigurationDescription.RollbackEnabled'
```
**Aktifkan auto-rollback (jika tidak diaktifkan)**
```
aws kinesisanalyticsv2 update-application \
--application-name MyApplication \
--current-application-version-id \
--application-configuration-update '{
"ApplicationSystemRollbackConfigurationUpdate": {
"RollbackEnabledUpdate": true
}
}'
```
## Fase 3: Ambil snapshot (opsional)
Jika snapshot otomatis diaktifkan untuk aplikasi Anda, Anda dapat melewati langkah ini, jika tidak, ambil snapshot aplikasi Anda untuk menyimpan status aplikasi Anda sebelum memutakhirkan.
**Ambil snapshot dari menjalankan aplikasi**
*Konsol Manajemen AWS:*
1. Arahkan ke aplikasi Anda
1. Pilih **Snapshots**
1. Pilih **Buat snapshot**
1. Masukkan nama snapshot (misalnya,`pre-flink-2.2-upgrade`)
1. Pilih **Buat**
*AWS CLI:*
```
aws kinesisanalyticsv2 create-application-snapshot \
--application-name MyApplication \
--snapshot-name pre-flink-2.2-upgrade
```
**Verifikasi pembuatan snapshot**
```
aws kinesisanalyticsv2 describe-application-snapshot \
--application-name MyApplication \
--snapshot-name pre-flink-2.2-upgrade
```
Tunggu sampai `SnapshotStatus` `READY` sebelum melanjutkan.
## Tahap 4: Tingkatkan aplikasi
Anda dapat meningkatkan aplikasi Flink Anda dengan menggunakan [https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)tindakan.
Anda dapat memanggil `UpdateApplication` API dengan berbagai cara:
+ **Gunakan Konsol Manajemen AWS.**
+ Buka halaman aplikasi Anda di file Konsol Manajemen AWS.
+ Pilih **Konfigurasikan**
+ Pilih runtime baru dan snapshot yang ingin Anda mulai, juga dikenal sebagai konfigurasi pemulihan. Gunakan pengaturan terbaru sebagai konfigurasi pemulihan untuk memulai aplikasi dari snapshot terbaru. Arahkan ke aplikasi baru yang ditingkatkan JAR/zip di Amazon S3.
+ **Gunakan AWS CLI[https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesisanalyticsv2/update-application.html](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesisanalyticsv2/update-application.html)**tindakan.
+ **Gunakan CloudFormation.**
+ Perbarui `RuntimeEnvironment` bidang. Sebelumnya, CloudFormation menghapus aplikasi dan membuat yang baru, menyebabkan snapshot Anda dan riwayat aplikasi lainnya hilang. Sekarang CloudFormation perbarui `RuntimeEnvironment` tempat Anda dan tidak menghapus aplikasi Anda.
+ **Gunakan AWS SDK.**
+ Konsultasikan dokumentasi SDK untuk bahasa pemrograman pilihan Anda. Lihat [https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html).
Anda dapat melakukan pemutakhiran saat aplikasi dalam `RUNNING` keadaan atau saat aplikasi dihentikan dalam `READY` status. Amazon Managed Service untuk Apache Flink memvalidasi kompatibilitas antara versi runtime asli dan versi runtime target. Pemeriksaan kompatibilitas ini berjalan saat Anda melakukan `UpdateApplication` saat dalam `RUNNING` status atau berikutnya `StartApplication` jika Anda memutakhirkan saat dalam `READY` status.
**Tingkatkan dari status RUNNING**
```
aws kinesisanalyticsv2 update-application \
--application-name MyApplication \
--current-application-version-id \
--runtime-environment-update FLINK-2_2 \
--application-configuration-update '{
"ApplicationCodeConfigurationUpdate": {
"CodeContentUpdate": {
"S3ContentLocationUpdate": {
"FileKeyUpdate": "my-app-flink-2.2.jar"
}
}
}
}'
```
**Tingkatkan dari status READY**
```
aws kinesisanalyticsv2 update-application \
--application-name MyApplication \
--current-application-version-id \
--runtime-environment-update FLINK-2_2 \
--application-configuration-update '{
"ApplicationCodeConfigurationUpdate": {
"CodeContentUpdate": {
"S3ContentLocationUpdate": {
"FileKeyUpdate": "my-app-flink-2.2.jar"
}
}
}
}'
```
## Tahap 5: Memantau peningkatan
**Pemeriksaan kompatibilitas**
+ Gunakan API Operasi untuk memeriksa status pemutakhiran. Jika ada ketidakcocokan biner atau masalah dengan startup pekerjaan, operasi pemutakhiran akan gagal dengan log.
+ Jika Operasi Upgrade telah berhasil tetapi aplikasi macet di loop restart, ini berarti status tidak kompatibel dengan versi Flink baru atau ada masalah dengan kode yang diperbarui. Tinjau [Panduan kompatibilitas status untuk peningkatan Flink 2.2](state-compatibility.md) tentang cara mengidentifikasi masalah ketidakcocokan negara.
**Pantau kesehatan aplikasi**
*Status aplikasi:*
+ Status aplikasi harus transisi: `RUNNING` → `UPDATING` → `RUNNING`
+ Periksa runtime aplikasi. Jika 2.2, operasi upgrade berhasil.
+ Jika aplikasi Anda dalam `RUNNING` tetapi masih pada runtime yang lebih lama, auto-rollback dimulai. Operasi API akan menampilkan operasi sebagai`FAILED`. Periksa log untuk menemukan pengecualian untuk kegagalan.
Selain itu, pantau metrik ini di CloudWatch:
*Mulai ulang metrik:*
+ `numRestarts`: Monitor untuk restart yang tidak terduga - upgrade berhasil `numRestarts` jika nol dan `uptime` atau `runningTime` meningkat.
*Metrik pos pemeriksaan:*
+ `lastCheckpointDuration`: Harus mirip dengan nilai pra-peningkatan
+ `numberOfFailedCheckpoints`: Harus tetap di 0
## Fase 6: Validasi perilaku aplikasi
Setelah aplikasi berjalan di Flink 2.2:
**Validasi fungsional**
+ Verifikasi data sedang dibaca dari sumber
+ Verifikasi data sedang ditulis ke sink
+ Verifikasi logika bisnis menghasilkan hasil yang diharapkan
+ Bandingkan output dengan baseline pra-upgrade
**Validasi kinerja**
+ Pantau metrik latensi (waktu end-to-end pemrosesan)
+ Pantau metrik throughput (catatan per detik)
+ Pantau durasi dan ukuran pos pemeriksaan
+ Memantau memori dan pemanfaatan CPU
**Jalankan selama 24\$1 jam**
Biarkan aplikasi berjalan setidaknya selama 24 jam dalam produksi untuk memastikan:
+ Tidak ada kebocoran memori
+ Perilaku pos pemeriksaan yang stabil
+ Tidak ada restart yang tidak terduga
+ Throughput yang konsisten
## Fase 7: Prosedur rollback
Jika pemutakhiran gagal atau aplikasi berjalan tetapi tidak sehat, putar kembali ke versi sebelumnya.
**Rollback otomatis**
Jika auto-rollback diaktifkan dan upgrade gagal selama startup, Amazon Managed Service untuk Apache Flink otomatis kembali ke versi sebelumnya.
**Rollback manual**
Jika aplikasi berjalan tetapi tidak sehat, gunakan `RollbackApplication` API:
*Konsol Manajemen AWS:*
1. Arahkan ke aplikasi Anda
1. Pilih **Tindakan** → **Gulung kembali**
1. Konfirmasikan rollback
*AWS CLI:*
```
aws kinesisanalyticsv2 rollback-application \
--application-name MyApplication \
--current-application-version-id
```
**Apa yang terjadi selama rollback:**
+ Aplikasi berhenti
+ Runtime kembali ke versi Flink sebelumnya
+ Kode aplikasi kembali ke JAR sebelumnya
+ Aplikasi dimulai ulang dari snapshot terakhir yang berhasil diambil **sebelum peningkatan**
**penting**
Anda tidak dapat memulihkan snapshot Flink 2.2 di Flink 1.x
Rollback menggunakan snapshot yang diambil sebelum upgrade
Selalu ambil snapshot sebelum meningkatkan (Fase 3)
## Langkah selanjutnya
Untuk pertanyaan atau masalah selama upgrade, lihat [Memecahkan Masalah Layanan Terkelola untuk Apache Flink](troubleshooting.md) atau hubungi AWS Support.
# Panduan kompatibilitas status untuk peningkatan Flink 2.2
Saat memutakhirkan dari Flink 1.x ke Flink 2.2, masalah kompatibilitas status dapat mencegah aplikasi Anda memulihkan dari snapshot. Panduan ini membantu Anda mengidentifikasi potensi masalah kompatibilitas dan menyediakan strategi migrasi.
## Memahami perubahan kompatibilitas status
Amazon Managed Service untuk Apache Flink 2.2 memperkenalkan beberapa perubahan serialisasi yang memengaruhi kompatibilitas status. Berikut ini adalah yang utama:
+ **Upgrade Versi Kryo**: Apache Flink 2.2 meningkatkan serializer Kryo yang dibundel dari versi 2 ke versi 5. Karena Kryo v5 menggunakan format pengkodean biner yang berbeda dari Kryo v2, status operator apa pun yang diserialkan melalui Kryo di savepoint Flink 1.x tidak dapat dipulihkan di Flink 2.2.
+ **Serialisasi Koleksi Java**: Di Flink 1.x, koleksi Java (seperti`HashMap`,`ArrayList`, dan`HashSet`) di dalamnya POJOs diserialkan menggunakan Kryo. Flink 2.2 memperkenalkan serializer yang dioptimalkan khusus koleksi yang tidak kompatibel dengan status serial KRYO dari 1.x. Aplikasi yang menggunakan koleksi Java dengan serializer POJO atau Kryo di 1.x tidak dapat mengembalikan status ini di Flink 2.2. Lihat [dokumentasi](https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/dev/datastream/fault-tolerance/serialization/types_serialization/) Flink untuk detail selengkapnya tentang tipe data dan serialisasi.
+ **Kompatibilitas Konektor Kinesis: Versi konektor** Kinesis Data Streams (KDS) yang lebih rendah dari 5.0 mempertahankan status yang tidak kompatibel dengan konektor Kinesis Flink 2.2 versi 6.0. Anda harus bermigrasi ke konektor versi 5.0 atau lebih tinggi sebelum upgrade Anda.
## Referensi kompatibilitas serialisasi
Tinjau semua deklarasi status dalam aplikasi Anda dan cocokkan jenis serialisasi dengan tabel di bawah ini. Jika ada jenis status yang tidak kompatibel, lihat [Migrasi negara](#state-compat-migration) bagian sebelum melanjutkan dengan pemutakhiran Anda.
**Referensi kompatibilitas serialisasi**
| Jenis Serialisasi | Kompatibel? | Detail |
| --- | --- | --- |
| Avro (SpecificRecord,GenericRecord) | Ya | Menggunakan format binernya sendiri yang independen dari Kryo. Pastikan Anda menggunakan informasi tipe Avro asli Flink, bukan Avro terdaftar sebagai serializer Kryo. |
| Protobuf | Ya | Menggunakan pengkodean binernya sendiri yang independen dari Kryo. Verifikasi perubahan skema mengikuti aturan evolusi yang kompatibel ke belakang. |
| POJOs tanpa koleksi | Ya | Ditangani oleh serializer POJO Flink — tetapi hanya jika kelas memenuhi semua kriteria POJO: kelas publik, konstruktor no-arg publik, semua bidang baik publik atau dapat diakses melalui getter/setter, dan semua jenis bidang itu sendiri dapat diserialkan oleh Flink. POJO yang melanggar salah satu dari ini diam-diam jatuh kembali ke Kryo dan menjadi tidak kompatibel. |
| Kustom TypeSerializers | Ya | Kompatibel hanya jika serializer Anda tidak mendelegasikan ke Kryo secara internal. |
| Status SQL dan Tabel API | Ya (dengan peringatan) | Menggunakan serializer internal Flink. Namun, Apache Flink tidak menjamin kompatibilitas status antara versi utama untuk aplikasi Table API. Uji di lingkungan non-produksi terlebih dahulu. |
| POJOs dengan koleksi Java (HashMap,ArrayList,HashSet) | Tidak | Di Flink 1.x, koleksi di dalamnya POJOs diserialkan melalui Kryo v2. Flink 2.2 memperkenalkan serializer koleksi khusus yang format binernya tidak kompatibel dengan format Kryo v2. |
| Kelas kasus scala | Tidak | Diserialkan melalui Kryo di Flink 1.x. Upgrade Kryo v2 ke v5 mengubah format biner. |
| Catatan Java | Tidak | Biasanya kembali ke serialisasi Kryo di Flink 1.x. Verifikasi dengan menguji dengandisableGenericTypes(). |
| Jenis pustaka pihak ketiga | Tidak | Jenis tanpa serializer kustom terdaftar kembali ke Kryo. Perubahan format biner Kryo v2 ke v5 merusak kompatibilitas. |
| Tipe apa pun menggunakan Kryo fallback | Tidak | Jika Flink tidak dapat menangani tipe dengan serializer bawaan atau terdaftar, itu kembali ke Kryo. Semua status serial Kryo dari 1.x tidak kompatibel dengan 2.2. |
## Metode diagnostik
Anda dapat mengidentifikasi masalah kompatibilitas status secara proaktif dengan melihat log aplikasi atau memeriksa log setelah operasi [UpdateApplication API](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html).
**Identifikasi Kryo fallback dalam aplikasi Anda**
Anda dapat menggunakan pola regex berikut di log Anda untuk mengidentifikasi fallback Kryo dalam aplikasi Anda:
```
Class class (?[^\s]+) cannot be used as a POJO type
```
Log sampel:
```
Class class org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber
cannot be used as a POJO type because not all fields are valid POJO fields,
and must be processed as GenericType. Please read the Flink documentation on
"Data Types & Serialization" for details of the effect on performance and
schema evolution.
```
Jika pemutakhiran gagal menggunakan UpdateApplication API, pengecualian berikut mungkin menandakan bahwa Anda mengalami ketidakcocokan status berbasis serializer:
**IndexOutOfBoundsException**
```
Caused by: java.lang.IndexOutOfBoundsException: Index 116 out of bounds for length 1
at java.base/jdk.internal.util.Preconditions.outOfBounds(Unknown Source)
at java.base/jdk.internal.util.Preconditions.outOfBoundsCheckIndex(Unknown Source)
at java.base/jdk.internal.util.Preconditions.checkIndex(Unknown Source)
at java.base/java.util.Objects.checkIndex(Unknown Source)
at java.base/java.util.ArrayList.get(Unknown Source)
at com.esotericsoftware.kryo.util.MapReferenceResolver.getReadObject(MapReferenceResolver.java:77)
at com.esotericsoftware.kryo.Kryo.readReferenceOrNull(Kryo.java:923)
... 23 more
```
**StateMigrationException (POJOSerializer)**
```
Caused by: org.apache.flink.util.StateMigrationException: The new state serializer
(org.apache.flink.api.java.typeutils.runtime.PojoSerializer@8bf85b5d) must not be
incompatible with the old state serializer
(org.apache.flink.api.java.typeutils.runtime.PojoSerializer@3282ee3).
```
## Daftar periksa pra-peningkatan
+ Tinjau semua deklarasi status dalam aplikasi Anda
+ Periksa POJOs dengan koleksi (`HashMap`,`ArrayList`,`HashSet`)
+ Verifikasi metode serialisasi untuk setiap jenis status
+ Buat aplikasi replika prod dan uji kompatibilitas status menggunakan UpdateApplication API pada replika ini
+ Jika status tidak kompatibel, pilih strategi dari [Migrasi negara](#state-compat-migration)
+ Aktifkan auto-rollback dalam konfigurasi aplikasi Flink produksi Anda
## Migrasi negara
**Membangun kembali status lengkap**
Terbaik untuk aplikasi di mana status dapat dibangun kembali dari data sumber.
Jika aplikasi Anda dapat membangun kembali status dari data sumber:
1. Hentikan aplikasi Flink 1.x
1. Tingkatkan ke Flink 2.x dengan kode yang diperbarui
1. Mulailah dengan `SKIP_RESTORE_FROM_SNAPSHOT`
1. Izinkan aplikasi untuk membangun kembali status
```
aws kinesisanalyticsv2 start-application \
--application-name MyApplication \
--run-configuration '{
"ApplicationRestoreConfiguration": {
"ApplicationRestoreType": "SKIP_RESTORE_FROM_SNAPSHOT"
}
}'
```
## Praktik terbaik
1. **Selalu gunakan Avro atau Protobuf untuk keadaan kompleks** — Ini memberikan evolusi skema dan tidak bergantung pada Kryo
1. **Hindari koleksi di POJOs** - Gunakan asli Flink `ListState` dan `MapState` sebagai gantinya
1. **Uji restorasi status secara lokal** - Sebelum peningkatan produksi, uji dengan snapshot aktual
1. **Sering-seringlah mengambil snapshot** — Terutama sebelum upgrade versi utama
1. **Aktifkan auto-rollback** - Konfigurasikan aplikasi MSF Anda untuk secara otomatis mengembalikan kegagalan
1. **Dokumentasikan jenis status Anda** — Pertahankan dokumentasi semua jenis status dan metode serialisasinya
1. **Pantau ukuran pos pemeriksaan - Ukuran** pos pemeriksaan yang berkembang dapat mengindikasikan masalah serialisasi
## Langkah selanjutnya
**Rencanakan peningkatan Anda**: Lihat[Upgrade ke Flink 2.2: Panduan lengkap](flink-2-2-upgrade-guide.md).
Untuk pertanyaan atau masalah selama migrasi, lihat [Memecahkan Masalah Layanan Terkelola untuk Apache Flink](troubleshooting.md) atau hubungi AWS Support.
# Menerapkan penskalaan aplikasi di Managed Service untuk Apache Flink
Anda dapat mengonfigurasi eksekusi tugas secara paralel dan alokasi sumber daya untuk Amazon Managed Service untuk Apache Flink untuk menerapkan penskalaan. Untuk informasi tentang cara Apache Flink menjadwalkan instance paralel tugas, [lihat Eksekusi Paralel](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/execution/parallel/) di Dokumentasi Apache Flink.
**Topics**
+ [Konfigurasikan paralelisme aplikasi dan KPU ParallelismPer](#how-parallelism)
+ [Alokasikan Unit Pengolahan Kinesis](#how-scaling-kpus)
+ [Perbarui paralelisme aplikasi Anda](#how-scaling-howto)
+ [Gunakan penskalaan otomatis di Managed Service untuk Apache Flink](how-scaling-auto.md)
+ [Pertimbangan MaxParallelism](#how-scaling-auto-max-parallelism)
## Konfigurasikan paralelisme aplikasi dan KPU ParallelismPer
Anda mengonfigurasi eksekusi paralel untuk tugas aplikasi Managed Service for Apache Flink (seperti membaca dari sumber atau mengeksekusi operator) menggunakan properti berikut: [https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ApplicationConfiguration.html](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ApplicationConfiguration.html)
+ `Parallelism` — Gunakan properti ini untuk mengatur paralelisme aplikasi Apache Flink default. Semua operator, sumber, dan sink mengeksekusi dengan paralelisme ini kecuali ditimpa dalam kode aplikasi. Default-nya adalah `1`, dan maksimum default adalah `256`.
+ `ParallelismPerKPU` — Gunakan properti ini untuk mengatur jumlah tugas paralel yang dapat dijadwalkan per Unit Pemrosesan Kinesis (KPU) aplikasi Anda. Default-nya adalah `1`, dan maksimumnya adalah `8`. Untuk aplikasi yang memiliki operasi pemblokiran (misalnya, I/O), nilai `ParallelismPerKPU` yang lebih tinggi mengarah pada penggunaan penuh sumber daya KPU.
**catatan**
Batas `Parallelism` untuk sama dengan `ParallelismPerKPU` kali batas untuk KPUs (yang memiliki default 64). KPUs Batas dapat ditingkatkan dengan meminta kenaikan batas. Untuk petunjuk tentang cara meminta peningkatan batas ini, lihat "Untuk meminta peningkatan batas" di [Service Quotas](https://docs.aws.amazon.com/general/latest/gr/aws_service_limits.html).
Untuk informasi tentang menyetel paralelisme tugas untuk operator tertentu, lihat [Menyetel Parallelism: Operator](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/execution/parallel/#operator-level) di Dokumentasi Apache Flink.
## Alokasikan Unit Pengolahan Kinesis
Layanan Terkelola untuk Apache Flink menyediakan kapasitas sebagai. KPUs Satu KPU memberi Anda 1 vCPU dan memori 4 GB. Untuk setiap KPU yang dialokasikan, penyimpanan aplikasi berjalan sebesar 50 GB juga disediakan.
Managed Service for Apache Flink menghitung KPUs yang diperlukan untuk menjalankan aplikasi Anda menggunakan `Parallelism` dan `ParallelismPerKPU` properti, sebagai berikut:
```
Allocated KPUs for the application = Parallelism/ParallelismPerKPU
```
Layanan Terkelola untuk Apache Flink dengan cepat memberikan sumber daya aplikasi Anda sebagai respons terhadap lonjakan throughput atau aktivitas pemrosesan. Ini akan menghapus sumber daya dari aplikasi Anda secara bertahap setelah lonjakan aktivitas telah berlalu. Untuk menonaktifkan alokasi otomatis sumber daya, atur nilai `AutoScalingEnabled` ke `false`, seperti yang dijelaskan nanti di [Perbarui paralelisme aplikasi Anda](#how-scaling-howto).
Batas default KPUs untuk aplikasi Anda adalah 64. Untuk petunjuk tentang cara meminta peningkatan batas ini, lihat "Untuk meminta peningkatan batas" di [Service Quotas](https://docs.aws.amazon.com/general/latest/gr/aws_service_limits.html).
**catatan**
KPU tambahan dikenakan biaya untuk keperluan orkestrasi. Untuk informasi selengkapnya, lihat [Layanan Terkelola untuk harga Apache Flink](https://aws.amazon.com/kinesis/data-analytics/pricing/).
## Perbarui paralelisme aplikasi Anda
Bagian ini berisi permintaan sampel untuk tindakan API yang mengatur paralelisme aplikasi. Untuk contoh dan petunjuk selengkapnya tentang cara menggunakan blok permintaan dengan tindakan API, lihat [Layanan Terkelola untuk kode contoh API Apache Flink](api-examples.md).
Permintaan contoh berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html) mengatur paralelisme saat Anda membuat aplikasi:
```
{
"ApplicationName": "string",
"RuntimeEnvironment":"FLINK-1_18",
"ServiceExecutionRole":"arn:aws:iam::123456789123:role/myrole",
"ApplicationConfiguration": {
"ApplicationCodeConfiguration":{
"CodeContent":{
"S3ContentLocation":{
"BucketARN":"arn:aws:s3:::amzn-s3-demo-bucket",
"FileKey":"myflink.jar",
"ObjectVersion":"AbCdEfGhIjKlMnOpQrStUvWxYz12345"
}
},
"CodeContentType":"ZIPFILE"
},
"FlinkApplicationConfiguration": {
"ParallelismConfiguration": {
"AutoScalingEnabled": "true",
"ConfigurationType": "CUSTOM",
"Parallelism": 4,
"ParallelismPerKPU": 4
}
}
}
}
```
Permintaan contoh berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) mengatur paralelisme untuk aplikasi yang sudah ada:
```
{
"ApplicationName": "MyApplication",
"CurrentApplicationVersionId": 4,
"ApplicationConfigurationUpdate": {
"FlinkApplicationConfigurationUpdate": {
"ParallelismConfigurationUpdate": {
"AutoScalingEnabledUpdate": "true",
"ConfigurationTypeUpdate": "CUSTOM",
"ParallelismPerKPUUpdate": 4,
"ParallelismUpdate": 4
}
}
}
}
```
Permintaan contoh berikut untuk tindakan [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) menonaktifkan paralelisme untuk aplikasi yang sudah ada:
```
{
"ApplicationName": "MyApplication",
"CurrentApplicationVersionId": 4,
"ApplicationConfigurationUpdate": {
"FlinkApplicationConfigurationUpdate": {
"ParallelismConfigurationUpdate": {
"AutoScalingEnabledUpdate": "false"
}
}
}
}
```
# Gunakan penskalaan otomatis di Managed Service untuk Apache Flink
Layanan Terkelola untuk Apache Flink secara elastis menskalakan paralelisme aplikasi Anda untuk mengakomodasi throughput data sumber Anda dan kompleksitas operator Anda untuk sebagian besar skenario. Penskalaan otomatis diaktifkan secara default. Layanan Terkelola untuk Apache Flink memantau penggunaan sumber daya (CPU) aplikasi Anda, dan secara elastis menskalakan paralelisme aplikasi Anda ke atas atau ke bawah:
+ Aplikasi Anda meningkatkan skala (meningkatkan paralelisme) jika CloudWatch metrik maksimum `containerCPUUtilization` lebih besar dari 75 persen atau lebih selama 15 menit. Itu berarti `ScaleUp` tindakan dimulai ketika ada 15 titik data berturut-turut dengan periode 1 menit sama dengan atau lebih dari 75 persen. Sebuah `ScaleUp` tindakan menggandakan `CurrentParallelism` aplikasi Anda. `ParallelismPerKPU`tidak dimodifikasi. Akibatnya, jumlah yang dialokasikan KPUs juga berlipat ganda.
+ Aplikasi Anda mengurangi (mengurangi paralelisme) ketika penggunaan CPU Anda tetap di bawah 10 persen selama enam jam. Itu berarti `ScaleDown` tindakan dimulai ketika ada 360 titik data berturut-turut dengan periode 1 menit kurang dari 10 persen. Sebuah `ScaleDown` tindakan membagi dua (membulatkan) paralelisme aplikasi. `ParallelismPerKPU`tidak dimodifikasi, dan jumlah yang dialokasikan KPUs juga menjadi dua (dibulatkan ke atas).
**catatan**
Maks periode `containerCPUUtilization` lebih dari 1 menit dapat direferensikan untuk menemukan korelasi dengan titik data yang digunakan untuk tindakan Penskalaan, tetapi tidak perlu untuk mencerminkan momen yang tepat ketika tindakan diinisialisasi.
Layanan Terkelola untuk Apache Flink tidak akan mengurangi `CurrentParallelism` nilai aplikasi Anda menjadi kurang dari pengaturan aplikasi Anda. `Parallelism`
Ketika Layanan Terkelola untuk layanan Apache Flink menskalakan aplikasi Anda, itu akan berada dalam status. `AUTOSCALING` Anda dapat memeriksa status aplikasi Anda saat ini menggunakan [ ListApplications](https://docs.aws.amazon.com//managed-flink/latest/apiv2/API_ListApplications.html)tindakan [ DescribeApplication](https://docs.aws.amazon.com//managed-flink/latest/apiv2/API_DescribeApplication.html)atau. Saat layanan menskalakan aplikasi Anda, satu-satunya tindakan API valid yang dapat Anda gunakan adalah [ StopApplication](https://docs.aws.amazon.com//managed-flink/latest/apiv2/API_ListApplications.html)dengan `Force` parameter yang disetel ke`true`.
Anda dapat menggunakan properti `AutoScalingEnabled` (bagian dari [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_FlinkApplicationConfiguration.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_FlinkApplicationConfiguration.html)) untuk mengaktifkan atau menonaktifkan perilaku penskalaan otomatis. AWS Akun Anda dikenakan biaya untuk KPUs Layanan Terkelola untuk ketentuan Apache Flink yang merupakan fungsi dari aplikasi `parallelism` dan `parallelismPerKPU` pengaturan Anda. Lonjakan aktivitas meningkatkan Layanan Terkelola Anda untuk biaya Apache Flink.
Untuk informasi tentang harga, lihat [Amazon Managed Service untuk harga Apache Flink](https://aws.amazon.com/kinesis/data-analytics/pricing/).
Perhatikan hal tentang penskalaan aplikasi berikut:
+ Penskalaan otomatis diaktifkan secara default.
+ Penskalaan tidak berlaku untuk notebook Studio. Namun, jika Anda men-deploy notebook Studio sebagai aplikasi dengan status tahan lama, penskalaan akan diterapkan ke aplikasi yang di-deploy.
+ Aplikasi Anda memiliki batas default 64 KPUs. Untuk informasi selengkapnya, lihat [Layanan Terkelola untuk kuota notebook Apache Flink dan Studio](limits.md).
+ Saat penskalaan otomatis memperbarui paralelisme aplikasi, aplikasi mengalami waktu henti. Untuk menghindari waktu henti ini, lakukan hal berikut:
+ Nonaktifkan penskalaan otomatis
+ Konfigurasikan aplikasi Anda `parallelism` dan `parallelismPerKPU` dengan [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)tindakan. Untuk informasi selengkapnya tentang menyetel setelan paralelisme aplikasi Anda, lihat. [Perbarui paralelisme aplikasi Anda](how-scaling.md#how-scaling-howto)
+ Pantau penggunaan sumber daya aplikasi Anda secara berkala untuk memverifikasi bahwa aplikasi Anda memiliki pengaturan paralelisme yang benar untuk beban kerjanya. Untuk informasi tentang pemantauan penggunaan sumber daya alokasi, lihat [Metrik dan dimensi dalam Layanan Terkelola untuk Apache Flink](metrics-dimensions.md).
## Menerapkan penskalaan otomatis khusus
Jika Anda menginginkan kontrol berbutir lebih halus pada penskalaan otomatis atau menggunakan metrik pemicu selain`containerCPUUtilization`, Anda dapat menggunakan contoh ini:
+ [AutoScaling](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/infrastructure/AutoScaling)
Contoh ini menggambarkan cara menskalakan Layanan Terkelola untuk aplikasi Apache Flink menggunakan CloudWatch metrik yang berbeda dari aplikasi Apache Flink, termasuk metrik dari Amazon MSK dan Amazon Kinesis Data Streams, yang digunakan sebagai sumber atau sink.
Untuk informasi tambahan, lihat [Pemantauan yang ditingkatkan dan penskalaan otomatis untuk Apache Flink](https://aws.amazon.com/blogs/big-data/enhanced-monitoring-and-automatic-scaling-for-apache-flink/).
## Menerapkan penskalaan otomatis terjadwal
Jika beban kerja Anda mengikuti profil yang dapat diprediksi dari waktu ke waktu, Anda mungkin lebih suka menskalakan aplikasi Apache Flink Anda terlebih dahulu. Ini menskalakan aplikasi Anda pada waktu yang dijadwalkan, sebagai lawan penskalaan secara reaktif berdasarkan metrik. Untuk mengatur penskalaan naik dan turun pada jam tetap dalam sehari, Anda dapat menggunakan contoh ini:
+ [ScheduledScaling](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/infrastructure/ScheduledScaling)
## Pertimbangan MaxParallelism
Paralelisme maksimum yang dapat diskalakan oleh pekerjaan Flink dibatasi oleh *minimum* `maxParallelism` di semua operator pekerjaan. Misalnya, jika Anda memiliki pekerjaan sederhana dengan hanya sumber dan wastafel, dan sumbernya memiliki `maxParallelism` 16 dan wastafel memiliki 8, aplikasi tidak dapat skala melampaui paralelisme 8.
Untuk mempelajari bagaimana default `maxParallelism` operator dihitung dan cara mengganti default, lihat [Mengatur Paralelisme Maksimum](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/execution/parallel/#setting-the-maximum-parallelism) dalam dokumentasi Apache Flink.
Sebagai aturan dasar, ketahuilah bahwa jika Anda tidak menentukan `maxParallelism` untuk operator mana pun dan Anda memulai aplikasi Anda dengan paralelisme kurang dari atau sama dengan 128, semua operator akan memiliki `maxParallelism` 128.
**catatan**
Paralelisme maksimum pekerjaan adalah batas atas paralelisme untuk penskalaan aplikasi Anda mempertahankan status.
Jika Anda `maxParallelism` memodifikasi aplikasi yang ada, aplikasi tidak akan dapat memulai ulang dari snapshot sebelumnya yang diambil dengan yang lama`maxParallelism`. Anda hanya dapat me-restart aplikasi tanpa snapshot.
Jika Anda berencana untuk menskalakan aplikasi Anda ke paralelisme yang lebih besar dari 128, Anda harus secara eksplisit mengatur dalam `maxParallelism` aplikasi Anda.
+ Logika penskalaan otomatis akan mencegah penskalaan pekerjaan Flink ke paralelisme yang akan melebihi paralelisme maksimum pekerjaan.
+ Jika Anda menggunakan penskalaan otomatis khusus atau penskalaan terjadwal, konfigurasikan agar tidak melebihi paralelisme maksimum pekerjaan.
+ Jika Anda secara manual menskalakan aplikasi Anda di luar paralelisme maksimum, aplikasi gagal untuk memulai.
# Tambahkan tag ke Layanan Terkelola untuk aplikasi Apache Flink
Bagian ini menjelaskan cara menambahkan tag metadata nilai kunci ke Layanan Terkelola untuk aplikasi Apache Flink. Tanda ini dapat digunakan untuk tujuan berikut:
+ Menentukan penagihan untuk Layanan Terkelola individual untuk aplikasi Apache Flink. Untuk informasi selengkapnya, lihat [Menggunakan Tanda Alokasi Biaya](https://docs.aws.amazon.com/awsaccountbilling/latest/aboutv2/cost-alloc-tags.html) dalam *Panduan Manajemen Penagihan dan Biaya*.
+ Mengontrol akses ke sumber daya aplikasi berdasarkan tanda. Untuk informasi selengkapnya, lihat [Mengontrol Akses Menggunakan Tanda](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_tags.html) di *Panduan Pengguna AWS Identity and Access Management *.
+ Tujuan yang ditentukan pengguna. Anda dapat menentukan fungsi aplikasi berdasarkan adanya tanda pengguna.
Catat informasi berikut tentang penandaan:
+ Jumlah maksimum tanda aplikasi termasuk tanda sistem. Jumlah maksimum tanda aplikasi yang ditentukan pengguna adalah 50.
+ Jika tindakan berisi daftar tanda yang memiliki nilai `Key` duplikat, layanan melempar `InvalidArgumentException`.
**Topics**
+ [Tambahkan tag saat aplikasi dibuat](how-tagging-create.md)
+ [Menambahkan atau memperbarui tag untuk aplikasi yang ada](how-tagging-add.md)
+ [Daftar tag untuk aplikasi](how-tagging-list.md)
+ [Hapus tag dari aplikasi](how-tagging-remove.md)
# Tambahkan tag saat aplikasi dibuat
Anda menambahkan tag saat membuat aplikasi menggunakan `tags` parameter [CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html)tindakan.
Contoh permintaan berikut menunjukkan simpul `Tags` untuk permintaan `CreateApplication`:
```
"Tags": [
{
"Key": "Key1",
"Value": "Value1"
},
{
"Key": "Key2",
"Value": "Value2"
}
]
```
# Menambahkan atau memperbarui tag untuk aplikasi yang ada
Anda menambahkan tag ke aplikasi menggunakan [TagResource](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_TagResource.html)tindakan. Anda tidak dapat menambahkan tag ke aplikasi menggunakan [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)tindakan.
Untuk memperbarui tanda yang ada, tambahkan tanda dengan kunci yang sama dengan tanda yang ada.
Contoh permintaan berikut untuk tindakan `TagResource` menambahkan tanda baru atau memperbarui tanda yang ada:
```
{
"ResourceARN": "string",
"Tags": [
{
"Key": "NewTagKey",
"Value": "NewTagValue"
},
{
"Key": "ExistingKeyOfTagToUpdate",
"Value": "NewValueForExistingTag"
}
]
}
```
# Daftar tag untuk aplikasi
Untuk mencantumkan tag yang ada, Anda menggunakan [ListTagsForResource](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ListTagsForResource.html)tindakan.
Contoh permintaan berikut untuk tindakan `ListTagsForResource` mencantumkan tanda untuk aplikasi:
```
{
"ResourceARN": "arn:aws:kinesisanalyticsus-west-2:012345678901:application/MyApplication"
}
```
# Hapus tag dari aplikasi
Untuk menghapus tag dari aplikasi, Anda menggunakan [UntagResource](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UntagResource.html)tindakan.
Contoh permintaan berikut untuk tindakan `UntagResource` menghapus tanda dari aplikasi:
```
{
"ResourceARN": "arn:aws:kinesisanalyticsus-west-2:012345678901:application/MyApplication",
"TagKeys": [ "KeyOfFirstTagToRemove", "KeyOfSecondTagToRemove" ]
}
```
# Gunakan CloudFormation dengan Managed Service untuk Apache Flink
Latihan berikut menunjukkan cara memulai aplikasi Flink yang dibuat dengan CloudFormation menggunakan fungsi Lambda di tumpukan yang sama.
## Sebelum Anda mulai
Sebelum Anda memulai latihan ini, ikuti langkah-langkah untuk membuat aplikasi Flink menggunakan CloudFormation at [AWS::KinesisAnalytics::Application](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-kinesis-analyticsapplication.html).
## Tulis fungsi Lambda
[Untuk memulai aplikasi Flink setelah pembuatan atau pembaruan, kami menggunakan kinesisanalyticsv2 start-application API.](https://docs.aws.amazon.com/cli/latest/reference/kinesisanalyticsv2/start-application.html) Panggilan akan dipicu oleh CloudFormation peristiwa setelah pembuatan aplikasi Flink. Kita akan membahas cara mengatur tumpukan untuk memicu fungsi Lambda nanti dalam latihan ini, tetapi pertama-tama kita fokus pada deklarasi fungsi Lambda dan kodenya. Kami menggunakan `Python3.8` runtime dalam contoh ini.
```
StartApplicationLambda:
Type: AWS::Lambda::Function
DependsOn: StartApplicationLambdaRole
Properties:
Description: Starts an application when invoked.
Runtime: python3.8
Role: !GetAtt StartApplicationLambdaRole.Arn
Handler: index.lambda_handler
Timeout: 30
Code:
ZipFile: |
import logging
import cfnresponse
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info('Incoming CFN event {}'.format(event))
try:
application_name = event['ResourceProperties']['ApplicationName']
# filter out events other than Create or Update,
# you can also omit Update in order to start an application on Create only.
if event['RequestType'] not in ["Create", "Update"]:
logger.info('No-op for Application {} because CFN RequestType {} is filtered'.format(application_name, event['RequestType']))
cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
return
# use kinesisanalyticsv2 API to start an application.
client_kda = boto3.client('kinesisanalyticsv2', region_name=event['ResourceProperties']['Region'])
# get application status.
describe_response = client_kda.describe_application(ApplicationName=application_name)
application_status = describe_response['ApplicationDetail']['ApplicationStatus']
# an application can be started from 'READY' status only.
if application_status != 'READY':
logger.info('No-op for Application {} because ApplicationStatus {} is filtered'.format(application_name, application_status))
cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
return
# create RunConfiguration.
run_configuration = {
'ApplicationRestoreConfiguration': {
'ApplicationRestoreType': 'RESTORE_FROM_LATEST_SNAPSHOT',
}
}
logger.info('RunConfiguration for Application {}: {}'.format(application_name, run_configuration))
# this call doesn't wait for an application to transfer to 'RUNNING' state.
client_kda.start_application(ApplicationName=application_name, RunConfiguration=run_configuration)
logger.info('Started Application: {}'.format(application_name))
cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
except Exception as err:
logger.error(err)
cfnresponse.send(event,context, cfnresponse.FAILED, {"Data": str(err)})
```
Dalam kode sebelumnya, Lambda memproses CloudFormation peristiwa yang masuk, menyaring semuanya selain `Create` dan`Update`, mendapatkan status aplikasi dan memulainya jika statusnya. `READY` Untuk mendapatkan status aplikasi, Anda harus membuat peran Lambda, seperti yang ditunjukkan berikut.
## Buat peran Lambda
Anda membuat peran agar Lambda berhasil “berbicara” dengan aplikasi dan menulis log. Peran ini menggunakan kebijakan terkelola default, tetapi Anda mungkin ingin mempersempitnya menjadi menggunakan kebijakan khusus.
```
StartApplicationLambdaRole:
Type: AWS::IAM::Role
DependsOn: TestFlinkApplication
Properties:
Description: A role for lambda to use while interacting with an application.
AssumeRolePolicyDocument:
Version: '2012-10-17 '
Statement:
- Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
Action:
- sts:AssumeRole
ManagedPolicyArns:
- arn:aws:iam::aws:policy/Amazonmanaged-flinkFullAccess
- arn:aws:iam::aws:policy/CloudWatchLogsFullAccess
Path: /
```
Perhatikan bahwa sumber daya Lambda akan dibuat setelah pembuatan aplikasi Flink di tumpukan yang sama karena mereka bergantung padanya.
## Panggil fungsi Lambda
Sekarang yang tersisa hanyalah memanggil fungsi Lambda. Anda melakukan ini dengan menggunakan sumber [daya khusus](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-cfn-customresource.html).
```
StartApplicationLambdaInvoke:
Description: Invokes StartApplicationLambda to start an application.
Type: AWS::CloudFormation::CustomResource
DependsOn: StartApplicationLambda
Version: "1.0"
Properties:
ServiceToken: !GetAtt StartApplicationLambda.Arn
Region: !Ref AWS::Region
ApplicationName: !Ref TestFlinkApplication
```
Ini semua yang Anda butuhkan untuk memulai aplikasi Flink Anda menggunakan Lambda. Anda sekarang siap untuk membuat tumpukan Anda sendiri atau menggunakan contoh lengkap di bawah ini untuk melihat bagaimana semua langkah tersebut bekerja dalam praktik.
## Tinjau contoh yang diperluas
Contoh berikut adalah versi yang sedikit diperpanjang dari langkah-langkah sebelumnya dengan `RunConfiguration` penyesuaian tambahan yang dilakukan melalui [parameter template](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/parameters-section-structure.html). Ini adalah tumpukan kerja untuk Anda coba. Pastikan untuk membaca catatan yang menyertainya:
tumpukan.yaml
```
Description: 'kinesisanalyticsv2 CloudFormation Test Application'
Parameters:
ApplicationRestoreType:
Description: ApplicationRestoreConfiguration option, can be SKIP_RESTORE_FROM_SNAPSHOT, RESTORE_FROM_LATEST_SNAPSHOT or RESTORE_FROM_CUSTOM_SNAPSHOT.
Type: String
Default: SKIP_RESTORE_FROM_SNAPSHOT
AllowedValues: [ SKIP_RESTORE_FROM_SNAPSHOT, RESTORE_FROM_LATEST_SNAPSHOT, RESTORE_FROM_CUSTOM_SNAPSHOT ]
SnapshotName:
Description: ApplicationRestoreConfiguration option, name of a snapshot to restore to, used with RESTORE_FROM_CUSTOM_SNAPSHOT ApplicationRestoreType.
Type: String
Default: ''
AllowNonRestoredState:
Description: FlinkRunConfiguration option, can be true or false.
Default: true
Type: String
AllowedValues: [ true, false ]
CodeContentBucketArn:
Description: ARN of a bucket with application code.
Type: String
CodeContentFileKey:
Description: A jar filename with an application code inside a bucket.
Type: String
Conditions:
IsSnapshotNameEmpty: !Equals [ !Ref SnapshotName, '' ]
Resources:
TestServiceExecutionRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17 '
Statement:
- Effect: Allow
Principal:
Service:
- kinesisanlaytics.amazonaws.com
Action: sts:AssumeRole
ManagedPolicyArns:
- arn:aws:iam::aws:policy/AmazonKinesisFullAccess
- arn:aws:iam::aws:policy/AmazonS3FullAccess
Path: /
InputKinesisStream:
Type: AWS::Kinesis::Stream
Properties:
ShardCount: 1
OutputKinesisStream:
Type: AWS::Kinesis::Stream
Properties:
ShardCount: 1
TestFlinkApplication:
Type: 'AWS::kinesisanalyticsv2::Application'
Properties:
ApplicationName: 'CFNTestFlinkApplication'
ApplicationDescription: 'Test Flink Application'
RuntimeEnvironment: 'FLINK-1_18'
ServiceExecutionRole: !GetAtt TestServiceExecutionRole.Arn
ApplicationConfiguration:
EnvironmentProperties:
PropertyGroups:
- PropertyGroupId: 'KinesisStreams'
PropertyMap:
INPUT_STREAM_NAME: !Ref InputKinesisStream
OUTPUT_STREAM_NAME: !Ref OutputKinesisStream
AWS_REGION: !Ref AWS::Region
FlinkApplicationConfiguration:
CheckpointConfiguration:
ConfigurationType: 'CUSTOM'
CheckpointingEnabled: True
CheckpointInterval: 1500
MinPauseBetweenCheckpoints: 500
MonitoringConfiguration:
ConfigurationType: 'CUSTOM'
MetricsLevel: 'APPLICATION'
LogLevel: 'INFO'
ParallelismConfiguration:
ConfigurationType: 'CUSTOM'
Parallelism: 1
ParallelismPerKPU: 1
AutoScalingEnabled: True
ApplicationSnapshotConfiguration:
SnapshotsEnabled: True
ApplicationCodeConfiguration:
CodeContent:
S3ContentLocation:
BucketARN: !Ref CodeContentBucketArn
FileKey: !Ref CodeContentFileKey
CodeContentType: 'ZIPFILE'
StartApplicationLambdaRole:
Type: AWS::IAM::Role
DependsOn: TestFlinkApplication
Properties:
Description: A role for lambda to use while interacting with an application.
AssumeRolePolicyDocument:
Version: '2012-10-17 '
Statement:
- Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
Action:
- sts:AssumeRole
ManagedPolicyArns:
- arn:aws:iam::aws:policy/Amazonmanaged-flinkFullAccess
- arn:aws:iam::aws:policy/CloudWatchLogsFullAccess
Path: /
StartApplicationLambda:
Type: AWS::Lambda::Function
DependsOn: StartApplicationLambdaRole
Properties:
Description: Starts an application when invoked.
Runtime: python3.8
Role: !GetAtt StartApplicationLambdaRole.Arn
Handler: index.lambda_handler
Timeout: 30
Code:
ZipFile: |
import logging
import cfnresponse
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info('Incoming CFN event {}'.format(event))
try:
application_name = event['ResourceProperties']['ApplicationName']
# filter out events other than Create or Update,
# you can also omit Update in order to start an application on Create only.
if event['RequestType'] not in ["Create", "Update"]:
logger.info('No-op for Application {} because CFN RequestType {} is filtered'.format(application_name, event['RequestType']))
cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
return
# use kinesisanalyticsv2 API to start an application.
client_kda = boto3.client('kinesisanalyticsv2', region_name=event['ResourceProperties']['Region'])
# get application status.
describe_response = client_kda.describe_application(ApplicationName=application_name)
application_status = describe_response['ApplicationDetail']['ApplicationStatus']
# an application can be started from 'READY' status only.
if application_status != 'READY':
logger.info('No-op for Application {} because ApplicationStatus {} is filtered'.format(application_name, application_status))
cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
return
# create RunConfiguration from passed parameters.
run_configuration = {
'FlinkRunConfiguration': {
'AllowNonRestoredState': event['ResourceProperties']['AllowNonRestoredState'] == 'true'
},
'ApplicationRestoreConfiguration': {
'ApplicationRestoreType': event['ResourceProperties']['ApplicationRestoreType'],
}
}
# add SnapshotName to RunConfiguration if specified.
if event['ResourceProperties']['SnapshotName'] != '':
run_configuration['ApplicationRestoreConfiguration']['SnapshotName'] = event['ResourceProperties']['SnapshotName']
logger.info('RunConfiguration for Application {}: {}'.format(application_name, run_configuration))
# this call doesn't wait for an application to transfer to 'RUNNING' state.
client_kda.start_application(ApplicationName=application_name, RunConfiguration=run_configuration)
logger.info('Started Application: {}'.format(application_name))
cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
except Exception as err:
logger.error(err)
cfnresponse.send(event,context, cfnresponse.FAILED, {"Data": str(err)})
StartApplicationLambdaInvoke:
Description: Invokes StartApplicationLambda to start an application.
Type: AWS::CloudFormation::CustomResource
DependsOn: StartApplicationLambda
Version: "1.0"
Properties:
ServiceToken: !GetAtt StartApplicationLambda.Arn
Region: !Ref AWS::Region
ApplicationName: !Ref TestFlinkApplication
ApplicationRestoreType: !Ref ApplicationRestoreType
SnapshotName: !Ref SnapshotName
AllowNonRestoredState: !Ref AllowNonRestoredState
```
Sekali lagi, Anda mungkin ingin menyesuaikan peran untuk Lambda serta aplikasi itu sendiri.
Sebelum membuat tumpukan di atas, jangan lupa untuk menentukan parameter Anda.
parameters.json
```
[
{
"ParameterKey": "CodeContentBucketArn",
"ParameterValue": "YOUR_BUCKET_ARN"
},
{
"ParameterKey": "CodeContentFileKey",
"ParameterValue": "YOUR_JAR"
},
{
"ParameterKey": "ApplicationRestoreType",
"ParameterValue": "SKIP_RESTORE_FROM_SNAPSHOT"
},
{
"ParameterKey": "AllowNonRestoredState",
"ParameterValue": "true"
}
]
```
Ganti `YOUR_BUCKET_ARN` dan `YOUR_JAR` dengan kebutuhan spesifik Anda. Anda dapat mengikuti [panduan](https://docs.aws.amazon.com/managed-flink/latest/java/get-started-exercise.html) ini untuk membuat ember Amazon S3 dan toples aplikasi.
Sekarang buat tumpukan (ganti YOUR\$1REGION dengan wilayah pilihan Anda, misalnya us-east-1):
```
aws cloudformation create-stack --region YOUR_REGION --template-body "file://stack.yaml" --parameters "file://parameters.json" --stack-name "TestManaged Service for Apache FlinkStack" --capabilities CAPABILITY_NAMED_IAM
```
Anda sekarang dapat menavigasi [https://console---aws.amazon.com.rproxy.govskope.cake/cloudformation](https://console.aws.amazon.com/cloudformation) dan melihat kemajuannya. Setelah dibuat, Anda akan melihat aplikasi Flink Anda dalam `Starting` keadaan. Mungkin perlu beberapa menit sampai dimulai`Running`.
Untuk informasi selengkapnya, lihat berikut ini:
+ [Empat cara untuk mengambil properti AWS layanan apa pun menggunakan AWS CloudFormation (Bagian 1 dari 3)](https://aws.amazon.com/blogs/mt/four-ways-to-retrieve-any-aws-service-property-using-aws-cloudformation-part-1/).
+ [Panduan: Mencari Gambar Mesin Amazon](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/walkthrough-custom-resources-lambda-lookup-amiids.html). IDs
# Gunakan Apache Flink Dashboard dengan Managed Service untuk Apache Flink
Anda dapat menggunakan Apache Flink Dashboard aplikasi Anda untuk memantau Layanan Terkelola Anda untuk kesehatan aplikasi Apache Flink. Dasbor aplikasi Anda menunjukkan informasi berikut:
+ Sumber daya yang digunakan, termasuk Manajer Tugas dan Slot Tugas.
+ Informasi tentang Tugas, termasuk yang berjalan, selesai, dibatalkan, dan gagal.
Untuk informasi tentang Manajer Tugas, Slot Tugas, dan Tugas Apache Flink, lihat [Arsitektur Apache Flink](https://flink.apache.org/what-is-flink/flink-architecture/) di situs web Apache Flink.
Perhatikan hal berikut tentang menggunakan Apache Flink Dashboard dengan Managed Service untuk aplikasi Apache Flink:
+ Dasbor Apache Flink untuk Layanan Terkelola untuk aplikasi Apache Flink hanya bisa dibaca. Anda tidak dapat membuat perubahan pada aplikasi Managed Service for Apache Flink menggunakan Apache Flink Dashboard.
+ Dasbor Apache Flink tidak kompatibel dengan Microsoft Internet Explorer.
## Akses Apache Flink Dashboard aplikasi Anda
Anda dapat mengakses Apache Flink Dashboard aplikasi Anda baik melalui Managed Service for Apache Flink console, atau dengan meminta endpoint URL aman menggunakan CLI.
### Mengakses Apache Flink Dashboard aplikasi Anda menggunakan Managed Service for Apache Flink console
Untuk mengakses Dasbor Apache Flink aplikasi Anda dari konsol, pilih **Apache Flink Dashboard** (Dasbor Apache Flink) di halaman aplikasi Anda.
**catatan**
Saat Anda membuka dasbor dari Layanan Terkelola untuk konsol Apache Flink, URL yang dihasilkan konsol akan berlaku selama 12 jam.
### Akses Apache Flink Dashboard aplikasi Anda menggunakan Managed Service for Apache Flink CLI
Anda dapat menggunakan Managed Service for Apache Flink CLI untuk menghasilkan URL untuk mengakses dasbor aplikasi Anda. URL yang Anda buat berlaku selama waktu tertentu.
**catatan**
Jika Anda tidak mengakses URL yang dibuat dalam waktu tiga menit, URL tersebut tidak akan berlaku lagi.
Anda membuat URL dasbor Anda menggunakan [ CreateApplicationPresignedUrl](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplicationPresignedUrl.html)tindakan. Anda menentukan parameter berikut untuk tindakan:
+ Nama aplikasi
+ Waktu dalam detik ketika URL akan valid
+ Anda menentukan `FLINK_DASHBOARD_URL` sebagai tipe URL.