Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Menulis data ke Amazon Kinesis Data Streams
*Produser* adalah aplikasi yang menulis data ke Amazon Kinesis Data Streams. Anda dapat membangun produsen untuk Kinesis Data Streams AWS SDK untuk Java menggunakan dan Kinesis Producer Library (KPL).
Jika Anda baru mengenal Kinesis Data Streams, mulailah dengan menjadi akrab dengan konsep dan terminologi [Apa itu Amazon Kinesis Data Streams?](introduction.md) yang disajikan dalam dan. [Gunakan AWS CLI untuk menjalankan operasi Amazon Kinesis Data Streams](getting-started.md)
**penting**
Kinesis Data Streams mendukung perubahan pada periode retensi rekaman data aliran data Anda. Untuk informasi selengkapnya, lihat [Ubah periode retensi data](kinesis-extended-retention.md).
Untuk memasukkan data ke dalam aliran, Anda harus menentukan nama aliran, kunci partisi, dan gumpalan data yang akan ditambahkan ke aliran. Kunci partisi digunakan untuk menentukan pecahan mana dalam aliran catatan data ditambahkan.
Semua data dalam pecahan dikirim ke pekerja yang sama yang memproses pecahan. Kunci partisi mana yang Anda gunakan tergantung pada logika aplikasi Anda. Jumlah kunci partisi biasanya harus jauh lebih besar daripada jumlah pecahan. Ini karena kunci partisi digunakan untuk menentukan cara memetakan catatan data ke pecahan tertentu. Jika Anda memiliki cukup kunci partisi, data dapat didistribusikan secara merata di seluruh pecahan dalam aliran.
**Topics**
+ [Kembangkan produsen menggunakan Amazon Kinesis Producer Library (KPL)](developing-producers-with-kpl.md)
+ [Kembangkan produsen menggunakan API Amazon Kinesis Data Streams dengan AWS SDK untuk Java](developing-producers-with-sdk.md)
+ [Menulis ke Amazon Kinesis Data Streams menggunakan Agen Kinesis](writing-with-agents.md)
+ [Menulis ke Kinesis Data AWS Streams menggunakan layanan lain](using-other-services.md)
+ [Menulis ke Kinesis Data Streams menggunakan integrasi pihak ketiga](using-other-services-third-party.md)
+ [Memecahkan masalah produsen Amazon Kinesis Data Streams](troubleshooting-producers.md)
+ [Optimalkan produsen Kinesis Data Streams](advanced-producers.md)
# Kembangkan produsen menggunakan Amazon Kinesis Producer Library (KPL)
*Produsen Amazon Kinesis Data Streams adalah aplikasi yang menempatkan catatan data pengguna ke dalam aliran data Kinesis (juga disebut konsumsi data).* Amazon Kinesis Producer Library (KPL) menyederhanakan pengembangan aplikasi produser, memungkinkan pengembang mencapai throughput penulisan yang tinggi ke aliran data Kinesis.
Anda dapat memantau KPL dengan Amazon CloudWatch. Untuk informasi selengkapnya, lihat [Pantau Perpustakaan Produser Kinesis dengan Amazon CloudWatch](monitoring-with-kpl.md).
**Topics**
+ [Tinjau peran KPL](#developing-producers-with-kpl-role)
+ [Sadarilah keuntungan menggunakan KPL](#developing-producers-with-kpl-advantage)
+ [Memahami kapan tidak menggunakan KPL](#developing-producers-with-kpl-when)
+ [Instal KPL](kinesis-kpl-dl-install.md)
+ [Migrasi dari KPL 0.x ke KPL 1.x](kpl-migration-1x.md)
+ [Transisi ke sertifikat Amazon Trust Services (ATS) untuk KPL](kinesis-kpl-upgrades.md)
+ [Platform yang didukung KPL](kinesis-kpl-supported-plats.md)
+ [Konsep kunci KPL](kinesis-kpl-concepts.md)
+ [Integrasikan KPL dengan kode produsen](kinesis-kpl-integration.md)
+ [Menulis ke aliran data Kinesis Anda menggunakan KPL](kinesis-kpl-writing.md)
+ [Konfigurasikan Perpustakaan Produsen Amazon Kinesis](kinesis-kpl-config.md)
+ [Menerapkan de-agregasi konsumen](kinesis-kpl-consumer-deaggregation.md)
+ [Gunakan KPL dengan Amazon Data Firehose](kpl-with-firehose.md)
+ [Gunakan KPL dengan AWS Glue Schema Registry](kpl-with-schemaregistry.md)
+ [Konfigurasikan konfigurasi proxy KPL](kpl-proxy-configuration.md)
+ [Kebijakan siklus hidup versi KPL](kpl-version-lifecycle-policy.md)
**catatan**
Disarankan agar Anda meningkatkan ke versi KPL terbaru. KPL diperbarui secara berkala dengan rilis yang lebih baru yang mencakup patch dependensi dan keamanan terbaru, perbaikan bug, dan fitur baru yang kompatibel ke belakang. Untuk informasi selengkapnya, lihat [https://github.com/awslabs/amazon-kinesis-producer/releases/](https://github.com/awslabs/amazon-kinesis-producer/releases/).
## Tinjau peran KPL
KPL adalah pustaka yang easy-to-use sangat dapat dikonfigurasi yang membantu Anda menulis ke aliran data Kinesis. Ini bertindak sebagai perantara antara kode aplikasi produser Anda dan tindakan API Kinesis Data Streams. KPL melakukan tugas-tugas utama berikut:
+ Menulis ke satu atau lebih aliran data Kinesis dengan mekanisme coba ulang otomatis dan dapat dikonfigurasi
+ Mengumpulkan catatan dan menggunakan `PutRecords` untuk menulis beberapa catatan ke beberapa pecahan per permintaan
+ Mengagregat catatan pengguna untuk meningkatkan ukuran payload dan meningkatkan throughput
+ Terintegrasi secara mulus dengan [Kinesis Client Library](https://docs.aws.amazon.com/kinesis/latest/dev/developing-consumers-with-kcl.html) (KCL) untuk melakukan de-agregasi batch record pada konsumen
+ Mengirimkan CloudWatch metrik Amazon atas nama Anda untuk memberikan visibilitas ke kinerja produsen
Perhatikan bahwa KPL berbeda dari Kinesis Data Streams API yang tersedia di. [AWS SDKs](https://aws.amazon.com/tools/) Kinesis Data Streams API membantu Anda mengelola banyak aspek Kinesis Data Streams (termasuk membuat stream, resharding, dan menempatkan dan mendapatkan record), sementara KPL menyediakan lapisan abstraksi khusus untuk menyerap data. Untuk informasi tentang Kinesis Data Streams API, lihat Referensi API [Amazon Kinesis](https://docs.aws.amazon.com/kinesis/latest/APIReference/).
## Sadarilah keuntungan menggunakan KPL
Daftar berikut merupakan beberapa keuntungan utama menggunakan KPL untuk mengembangkan produsen Kinesis Data Streams.
KPL dapat digunakan dalam kasus penggunaan sinkron atau asinkron. Kami menyarankan untuk menggunakan kinerja antarmuka asinkron yang lebih tinggi kecuali ada alasan khusus untuk menggunakan perilaku sinkron. Untuk informasi selengkapnya tentang dua kasus penggunaan dan kode contoh ini, lihat[Menulis ke aliran data Kinesis Anda menggunakan KPL](kinesis-kpl-writing.md).
**Manfaat Kinerja**
KPL dapat membantu membangun produsen berkinerja tinggi. Pertimbangkan situasi di mana instans Amazon EC2 Anda berfungsi sebagai proxy untuk mengumpulkan peristiwa 100 byte dari ratusan atau ribuan perangkat berdaya rendah dan menulis catatan ke dalam aliran data Kinesis. Instans EC2 ini masing-masing harus menulis ribuan peristiwa per detik ke aliran data Anda. Untuk mencapai throughput yang dibutuhkan, produsen harus menerapkan logika yang rumit, seperti batching atau multithreading, selain logika coba ulang dan de-agregasi rekaman di sisi konsumen. KPL melakukan semua tugas ini untuk Anda.
**Kemudahan Penggunaan Sisi Konsumen**
Untuk pengembang sisi konsumen yang menggunakan KCL di Jawa, KPL terintegrasi tanpa usaha tambahan. Ketika KCL mengambil rekaman Kinesis Data Streams agregat yang terdiri dari beberapa catatan pengguna KPL, secara otomatis memanggil KPL untuk mengekstrak catatan pengguna individu sebelum mengembalikannya ke pengguna.
Untuk pengembang sisi konsumen yang tidak menggunakan KCL tetapi menggunakan operasi API `GetRecords` secara langsung, perpustakaan Java KPL tersedia untuk mengekstrak catatan pengguna individu sebelum mengembalikannya ke pengguna.
**Pemantauan Produsen**
Anda dapat mengumpulkan, memantau, dan menganalisis produsen Kinesis Data Streams menggunakan CloudWatch Amazon dan KPL. KPL memancarkan throughput, kesalahan, dan metrik lainnya untuk CloudWatch atas nama Anda, dan dapat dikonfigurasi untuk memantau pada tingkat aliran, pecahan, atau produsen.
**Arsitektur Asinkron**
Karena KPL dapat menyangga catatan sebelum mengirimnya ke Kinesis Data Streams, itu tidak memaksa aplikasi pemanggil untuk memblokir dan menunggu konfirmasi bahwa catatan telah tiba di server sebelum melanjutkan runtime. Panggilan untuk memasukkan catatan ke dalam KPL selalu segera kembali dan tidak menunggu catatan dikirim atau respons diterima dari server. Sebaliknya, `Future` objek dibuat yang menerima hasil pengiriman catatan ke Kinesis Data Streams di lain waktu. Ini adalah perilaku yang sama dengan klien asinkron di SDK. AWS
## Memahami kapan tidak menggunakan KPL
KPL dapat menimbulkan penundaan pemrosesan tambahan hingga di `RecordMaxBufferedTime` dalam perpustakaan (dapat dikonfigurasi pengguna). Nilai yang lebih besar `RecordMaxBufferedTime` menghasilkan efisiensi pengepakan yang lebih tinggi dan kinerja yang lebih baik. Aplikasi yang tidak dapat mentolerir penundaan tambahan ini mungkin perlu menggunakan AWS SDK secara langsung. Untuk informasi selengkapnya tentang penggunaan AWS SDK dengan Kinesis Data Streams, lihat. [Kembangkan produsen menggunakan API Amazon Kinesis Data Streams dengan AWS SDK untuk Java](developing-producers-with-sdk.md) Untuk informasi selengkapnya tentang `RecordMaxBufferedTime` dan properti KPL yang dapat dikonfigurasi pengguna lainnya, lihat. [Konfigurasikan Perpustakaan Produsen Amazon Kinesis](kinesis-kpl-config.md)
# Instal KPL
Amazon menyediakan binari pra-bangun dari C\$1\$1 Amazon Kinesis Producer Library (KPL) untuk macOS, Windows, dan distribusi Linux terbaru (untuk detail platform yang didukung, lihat bagian selanjutnya). Binari ini dikemas sebagai bagian dari file Java .jar dan secara otomatis dipanggil dan digunakan jika Anda menggunakan Maven untuk menginstal paket. Untuk menemukan versi terbaru KPL dan KCL, gunakan tautan pencarian Maven berikut:
+ [KPL](https://search.maven.org/#search|ga|1|amazon-kinesis-producer)
+ [KCL](https://search.maven.org/#search|ga|1|amazon-kinesis-client)
Binari Linux telah dikompilasi dengan GNU Compiler Collection (GCC) dan ditautkan secara statis terhadap libstdc\$1\$1 di Linux. Mereka diharapkan untuk bekerja pada distribusi Linux 64-bit yang mencakup glibc versi 2.5 atau lebih tinggi.
Pengguna distribusi Linux sebelumnya dapat membangun KPL menggunakan instruksi build yang disediakan bersama dengan sumbernya. GitHub Untuk mengunduh KPL dari GitHub, lihat Perpustakaan [Produser Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-producer).
**penting**
Amazon Kinesis Producer Library (KPL) 0.x akan mencapai end-of-support pada 30 Januari 2026. Kami **sangat menyarankan agar** Anda memigrasikan aplikasi KPL Anda menggunakan versi 0.x ke versi KPL terbaru sebelum 30 Januari 2026. Untuk menemukan versi KPL terbaru, lihat [halaman KPL](https://github.com/awslabs/amazon-kinesis-producer) di Github. Untuk informasi tentang migrasi dari KPL 0.x ke KPL 1.x, lihat. [Migrasi dari KPL 0.x ke KPL 1.x](kpl-migration-1x.md)
# Migrasi dari KPL 0.x ke KPL 1.x
Topik ini memberikan step-by-step petunjuk untuk memigrasikan konsumen Anda dari KPL 0.x ke KPL 1.x. KPL 1.x memperkenalkan dukungan untuk AWS SDK untuk Java 2.x sambil mempertahankan kompatibilitas antarmuka dengan versi sebelumnya. Anda tidak perlu memperbarui logika pemrosesan data inti Anda untuk bermigrasi ke KPL 1.x.
1. **Pastikan Anda memiliki prasyarat berikut:**
+ Java Development Kit (JDK) 8 atau lebih baru
+ AWS SDK untuk Java 2.x
+ Maven atau Gradle untuk manajemen ketergantungan
1. **Tambahkan dependensi**
Jika Anda menggunakan Maven, tambahkan dependensi berikut ke file pom.xml Anda. Pastikan Anda memperbarui GroupId dari `com.amazonaws` ke `software.amazon.kinesis` dan versi `1.x.x` ke versi KPL terbaru.
```
software.amazon.kinesis
amazon-kinesis-producer
1.x.x
```
Jika Anda menggunakan Gradle, tambahkan berikut ini ke `build.gradle` file Anda. Pastikan untuk mengganti `1.x.x` dengan versi KPL terbaru.
```
implementation 'software.amazon.kinesis:amazon-kinesis-producer:1.x.x'
```
Anda dapat memeriksa versi terbaru KPL di Repositori Pusat [Maven](https://central.sonatype.com/search?q=amazon-kinesis-producer).
1. **Perbarui pernyataan impor untuk KPL**
KPL 1.x menggunakan AWS SDK untuk Java 2.x dan menggunakan nama paket yang diperbarui yang dimulai dengan`software.amazon.kinesis`, dibandingkan dengan nama paket di KPL sebelumnya yang dimulai dengan. `com.amazonaws.services.kinesis`
Ganti impor untuk `com.amazonaws.services.kinesis` dengan`software.amazon.kinesis`. Tabel berikut mencantumkan impor yang harus Anda ganti.
**Penggantian impor**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/id_id/streams/latest/dev/kpl-migration-1x.html)
1. **Perbarui pernyataan impor untuk kelas AWS penyedia kredensional**
Saat bermigrasi ke KPL 1.x, Anda harus memperbarui paket dan kelas dalam impor Anda dalam kode aplikasi KPL Anda yang didasarkan pada AWS SDK untuk Java 1.x ke yang sesuai berdasarkan 2.x. AWS SDK untuk Java Impor umum dalam aplikasi KPL adalah kelas penyedia kredensial. Lihat [Perubahan penyedia kredensial](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) dalam dokumentasi panduan migrasi AWS SDK untuk Java 2.x untuk mengetahui daftar lengkap perubahan penyedia kredensial. Berikut adalah perubahan impor umum yang mungkin perlu Anda lakukan di aplikasi KPL Anda.
**Impor di KPL 0.x**
```
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
```
**Impor di KPL 1.x**
```
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
```
Jika Anda mengimpor penyedia kredensi lain berdasarkan AWS SDK untuk Java 1.x, Anda harus memperbaruinya ke yang setara AWS SDK untuk Java 2.x. Jika Anda tidak mengimpor apa pun classes/packages dari AWS SDK untuk Java 1.x, Anda dapat mengabaikan langkah ini.
1. **Perbarui konfigurasi penyedia kredensional dalam konfigurasi KPL**
Konfigurasi penyedia kredensional di KPL 1.x memerlukan penyedia kredensi 2.x. AWS SDK untuk Java Jika Anda meneruskan penyedia kredensi untuk AWS SDK untuk Java 1.x di `KinesisProducerConfiguration` dengan mengganti penyedia kredensil default, Anda harus memperbaruinya dengan penyedia kredensi 2.x. AWS SDK untuk Java Lihat [Perubahan penyedia kredensial](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) dalam dokumentasi panduan migrasi AWS SDK untuk Java 2.x untuk mengetahui daftar lengkap perubahan penyedia kredensial. Jika Anda tidak mengganti penyedia kredensi default dalam konfigurasi KPL, Anda dapat mengabaikan langkah ini.
Misalnya, jika Anda mengganti penyedia kredensi default untuk KPL dengan kode berikut:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// SDK v1 default credentials provider
config.setCredentialsProvider(new DefaultAWSCredentialsProviderChain());
```
Anda harus memperbaruinya dengan kode berikut untuk menggunakan penyedia kredensi AWS SDK untuk Java 2.x:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// New SDK v2 default credentials provider
config.setCredentialsProvider(DefaultCredentialsProvider.create());
```
# Transisi ke sertifikat Amazon Trust Services (ATS) untuk KPL
Pada tanggal 9 Februari 2018, pukul 09.00 PST, Amazon Kinesis Data Streams memasang sertifikat ATS. [Untuk terus dapat menulis catatan ke Kinesis Data Streams menggunakan Amazon Kinesis Producer Library (KPL), Anda harus meng-upgrade instalasi KPL Anda ke versi 0.12.6 atau yang lebih baru.](http://search.maven.org/#artifactdetails|com.amazonaws|amazon-kinesis-producer|0.12.6|jar) Perubahan ini mempengaruhi semua AWS Wilayah.
Untuk informasi tentang perpindahan ke ATS, lihat [Cara AWS Mempersiapkan Pindah ke Otoritas Sertifikat Sendiri](https://aws.amazon.com/blogs/security/how-to-prepare-for-aws-move-to-its-own-certificate-authority/).
Jika Anda mengalami masalah dan membutuhkan dukungan teknis, [buat kasus](https://console.aws.amazon.com/support/v1#/case/create) dengan AWS Support Center.
# Platform yang didukung KPL
Amazon Kinesis Producer Library (KPL) ditulis dalam C\$1\$1 dan berjalan sebagai proses turunan ke proses pengguna utama. Binari asli 64-bit yang dikompilasi dibundel dengan rilis Java dan dikelola oleh pembungkus Java.
Paket Java berjalan tanpa perlu menginstal pustaka tambahan pada sistem operasi berikut:
+ Distribusi Linux dengan kernel 2.6.18 (September 2006) dan kemudian
+ Apple iOS X 10.9 dan yang lebih baru
+ Windows Server 2008 dan yang lebih baru
**penting**
Windows Server 2008 dan yang lebih baru didukung untuk semua versi KPL hingga versi 0.14.0.
Platform Windows TIDAK didukung dimulai dengan KPL versi 0.14.0 atau lebih tinggi.
Perhatikan bahwa KPL hanya 64-bit.
## Kode sumber
Jika binari yang disediakan dalam instalasi KPL tidak cukup untuk lingkungan Anda, inti KPL ditulis sebagai modul C\$1\$1. Kode sumber untuk modul C\$1\$1 dan antarmuka Java dirilis di bawah Lisensi Publik Amazon dan tersedia GitHub di Perpustakaan Produsen [Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-producer). Meskipun KPL dapat digunakan pada platform apa pun di mana kompiler C\$1\$1 dan JRE yang sesuai standar baru-baru ini tersedia, Amazon tidak secara resmi mendukung platform apa pun yang tidak ada dalam daftar platform yang didukung.
# Konsep kunci KPL
Bagian berikut berisi konsep dan terminologi yang diperlukan untuk memahami dan mendapatkan manfaat dari Amazon Kinesis Producer Library (KPL).
**Topics**
+ [Catatan](#kinesis-kpl-concepts-records)
+ [Batching](#kinesis-kpl-concepts-batching)
+ [Agregasi](#kinesis-kpl-concepts-aggretation)
+ [Koleksi](#kinesis-kpl-concepts-collection)
## Catatan
Dalam panduan ini, kami membedakan antara *catatan pengguna KPL dan catatan* *Kinesis* Data Streams. Saat kami menggunakan *catatan* istilah tanpa kualifikasi, kami merujuk ke catatan *pengguna KPL*. Ketika kami merujuk ke catatan Kinesis Data Streams, kami *secara eksplisit mengatakan catatan Kinesis* Data Streams.
Catatan pengguna KPL adalah gumpalan data yang memiliki arti khusus bagi pengguna. Contohnya termasuk gumpalan JSON yang mewakili peristiwa UI di situs web, atau entri log dari server web.
Catatan Kinesis Data Streams adalah instance `Record` dari struktur data yang ditentukan oleh API layanan Kinesis Data Streams. Ini berisi kunci partisi, nomor urut, dan gumpalan data.
## Batching
*Batching* mengacu pada melakukan satu tindakan pada beberapa item alih-alih berulang kali melakukan tindakan pada setiap item individu.
Dalam konteks ini, “item” adalah catatan, dan tindakan mengirimkannya ke Kinesis Data Streams. Dalam situasi non-batching, Anda akan menempatkan setiap rekaman dalam catatan Kinesis Data Streams terpisah dan membuat satu permintaan HTTP untuk mengirimkannya ke Kinesis Data Streams. Dengan batching, setiap permintaan HTTP dapat membawa beberapa catatan, bukan hanya satu.
KPL mendukung dua jenis batching:
+ *Agregasi* — Menyimpan beberapa catatan dalam satu catatan Kinesis Data Streams.
+ *Pengumpulan* — Menggunakan operasi API `PutRecords` untuk mengirim beberapa catatan Kinesis Data Streams ke satu atau beberapa pecahan dalam aliran data Kinesis Anda.
Kedua jenis batching KPL dirancang untuk hidup berdampingan dan dapat dihidupkan atau dimatikan secara independen satu sama lain. Secara default, keduanya dihidupkan.
## Agregasi
*Agregasi* mengacu pada penyimpanan beberapa catatan dalam catatan Kinesis Data Streams. Agregasi memungkinkan pelanggan untuk meningkatkan jumlah catatan yang dikirim per panggilan API, yang secara efektif meningkatkan throughput produsen.
Pecahan Kinesis Data Streams mendukung hingga 1.000 rekaman Kinesis Data Streams per detik, atau throughput 1 MB. Catatan Kinesis Data Streams per detik mengikat pelanggan dengan catatan yang lebih kecil dari 1 KB. Agregasi rekaman memungkinkan pelanggan untuk menggabungkan beberapa catatan ke dalam satu catatan Kinesis Data Streams. Hal ini memungkinkan pelanggan untuk meningkatkan throughput per shard mereka.
Pertimbangkan kasus satu pecahan di Wilayah us-east-1 yang saat ini berjalan pada tingkat konstan 1.000 catatan per detik, dengan catatan yang masing-masing 512 byte. Dengan agregasi KPL, Anda dapat mengemas 1.000 catatan hanya ke dalam 10 catatan Kinesis Data Streams, mengurangi RPS menjadi 10 (masing-masing 50 KB).
## Koleksi
*Collection* mengacu pada pengelompokan beberapa data Kinesis Data Streams dan mengirimkannya dalam satu permintaan HTTP dengan panggilan ke `PutRecords` operasi API, alih-alih mengirim setiap catatan Kinesis Data Streams dalam permintaan HTTP-nya sendiri.
Ini meningkatkan throughput dibandingkan dengan tidak menggunakan koleksi karena mengurangi overhead membuat banyak permintaan HTTP terpisah. Bahkan, `PutRecords` dirinya sendiri dirancang khusus untuk tujuan ini.
Koleksi berbeda dari agregasi karena bekerja dengan kelompok catatan Kinesis Data Streams. Rekaman Kinesis Data Streams yang dikumpulkan masih dapat berisi beberapa catatan dari pengguna. Hubungan tersebut dapat divisualisasikan seperti:
```
record 0 --|
record 1 | [ Aggregation ]
... |--> Amazon Kinesis record 0 --|
... | |
record A --| |
|
... ... |
|
record K --| |
record L | | [ Collection ]
... |--> Amazon Kinesis record C --|--> PutRecords Request
... | |
record S --| |
|
... ... |
|
record AA--| |
record BB | |
... |--> Amazon Kinesis record M --|
... |
record ZZ--|
```
# Integrasikan KPL dengan kode produsen
Amazon Kinesis Producer Library (KPL) berjalan dalam proses terpisah, dan berkomunikasi dengan proses pengguna induk Anda menggunakan IPC. Arsitektur ini kadang-kadang disebut [microservice](http://en.wikipedia.org/wiki/Microservices), dan dipilih karena dua alasan utama:
**1) Proses pengguna Anda tidak akan macet bahkan jika KPL mogok**
Proses Anda dapat memiliki tugas yang tidak terkait dengan Kinesis Data Streams, dan mungkin dapat melanjutkan operasi bahkan jika KPL mogok. Dimungkinkan juga bagi proses pengguna induk Anda untuk memulai ulang KPL dan memulihkan ke keadaan berfungsi penuh (fungsi ini ada di pembungkus resmi).
Contohnya adalah server web yang mengirimkan metrik ke Kinesis Data Streams; server dapat terus melayani halaman bahkan jika bagian Kinesis Data Streams telah berhenti bekerja. Merusak seluruh server karena bug di KPL akan menyebabkan pemadaman yang tidak perlu.
**2) Klien arbitrer dapat didukung**
Selalu ada pelanggan yang menggunakan bahasa selain yang didukung secara resmi. Pelanggan ini juga harus dapat menggunakan KPL dengan mudah.
## Matriks penggunaan yang disarankan
Matriks penggunaan berikut mencantumkan pengaturan yang disarankan untuk pengguna yang berbeda dan memberi tahu Anda tentang apakah dan bagaimana Anda harus menggunakan KPL. Perlu diingat bahwa jika agregasi diaktifkan, de-agregasi juga harus digunakan untuk mengekstrak catatan Anda di sisi konsumen.
| Bahasa sampingan produser | Bahasa sisi konsumen | Versi KCL | Logika titik pemeriksaan | Bisakah Anda menggunakan KPL? | Peringatan |
| --- | --- | --- | --- | --- | --- |
| Apa pun kecuali Java | \$1 | \$1 | \$1 | Tidak | N/A |
| Java | Java | Menggunakan Java SDK secara langsung | N/A | Ya | Jika agregasi digunakan, Anda harus menggunakan pustaka de-agregasi yang disediakan setelah panggilan. GetRecords |
| Java | Apa pun kecuali Java | Menggunakan SDK secara langsung | N/A | Ya | Harus menonaktifkan agregasi. |
| Java | Java | 1.3.x | N/A | Ya | Harus menonaktifkan agregasi. |
| Java | Java | 1.4.x | Panggilan pos pemeriksaan tanpa argumen | Ya | Tidak ada |
| Java | Java | 1.4.x | Memanggil pos pemeriksaan dengan nomor urut eksplisit | Ya | Nonaktifkan agregasi, atau ubah kode untuk menggunakan nomor urutan yang diperluas untuk pos pemeriksaan. |
| Java | Apa pun kecuali Java | 1.3.x\$1daemon multibahasa\$1pembungkus khusus bahasa | N/A | Ya | Harus menonaktifkan agregasi. |
# Menulis ke aliran data Kinesis Anda menggunakan KPL
Bagian berikut menunjukkan kode sampel dalam perkembangan dari produsen paling dasar ke kode asinkron sepenuhnya.
## Kode produsen Barebones
Kode berikut adalah semua yang diperlukan untuk menulis produsen kerja minimal. Catatan pengguna Amazon Kinesis Producer Library (KPL) diproses di latar belakang.
```
// KinesisProducer gets credentials automatically like
// DefaultAWSCredentialsProviderChain.
// It also gets region automatically from the EC2 metadata service.
KinesisProducer kinesis = new KinesisProducer();
// Put some records
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
kinesis.addUserRecord("myStream", "myPartitionKey", data);
}
// Do other stuff ...
```
## Menanggapi hasil secara sinkron
Pada contoh sebelumnya, kode tidak memeriksa apakah catatan pengguna KPL berhasil. KPL melakukan percobaan ulang yang diperlukan untuk memperhitungkan kegagalan. Tetapi jika Anda ingin memeriksa hasilnya, Anda dapat memeriksanya menggunakan `Future` objek yang dikembalikan`addUserRecord`, seperti pada contoh berikut (contoh sebelumnya ditampilkan untuk konteks):
```
KinesisProducer kinesis = new KinesisProducer();
// Put some records and save the Futures
List> putFutures = new LinkedList>();
for (int i = 0; i < 100; i++) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
putFutures.add(
kinesis.addUserRecord("myStream", "myPartitionKey", data));
}
// Wait for puts to finish and check the results
for (Future f : putFutures) {
UserRecordResult result = f.get(); // this does block
if (result.isSuccessful()) {
System.out.println("Put record into shard " +
result.getShardId());
} else {
for (Attempt attempt : result.getAttempts()) {
// Analyze and respond to the failure
}
}
}
```
## Menanggapi hasil secara asinkron
Contoh sebelumnya adalah memanggil `get()` `Future` objek, yang memblokir runtime. Jika Anda tidak ingin memblokir runtime, Anda dapat menggunakan callback asinkron, seperti yang ditunjukkan pada contoh berikut:
```
KinesisProducer kinesis = new KinesisProducer();
FutureCallback myCallback = new FutureCallback() {
@Override public void onFailure(Throwable t) {
/* Analyze and respond to the failure */
};
@Override public void onSuccess(UserRecordResult result) {
/* Respond to the success */
};
};
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
ListenableFuture f = kinesis.addUserRecord("myStream", "myPartitionKey", data);
// If the Future is complete by the time we call addCallback, the callback will be invoked immediately.
Futures.addCallback(f, myCallback);
}
```
# Konfigurasikan Perpustakaan Produsen Amazon Kinesis
Meskipun pengaturan default harus berfungsi dengan baik untuk sebagian besar kasus penggunaan, Anda mungkin ingin mengubah beberapa pengaturan default untuk menyesuaikan perilaku dengan `KinesisProducer` kebutuhan Anda. Sebuah instance dari `KinesisProducerConfiguration` kelas dapat diteruskan ke `KinesisProducer` konstruktor untuk melakukannya, misalnya:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration()
.setRecordMaxBufferedTime(3000)
.setMaxConnections(1)
.setRequestTimeout(60000)
.setRegion("us-west-1");
final KinesisProducer kinesisProducer = new KinesisProducer(config);
```
Anda juga dapat memuat konfigurasi dari file properti:
```
KinesisProducerConfiguration config = KinesisProducerConfiguration.fromPropertiesFile("default_config.properties");
```
Anda dapat mengganti jalur dan nama file apa pun yang dapat diakses oleh proses pengguna. Anda juga dapat memanggil metode set pada `KinesisProducerConfiguration` instance yang dibuat dengan cara ini untuk menyesuaikan konfigurasi.
File properti harus menentukan parameter menggunakan nama mereka di PascalCase. Nama-nama cocok dengan yang digunakan dalam metode set di `KinesisProducerConfiguration` kelas. Contoh:
```
RecordMaxBufferedTime = 100
MaxConnections = 4
RequestTimeout = 6000
Region = us-west-1
```
Untuk informasi selengkapnya tentang aturan penggunaan parameter konfigurasi dan batas nilai, lihat [contoh berkas properti konfigurasi GitHub](https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties).
Perhatikan bahwa setelah `KinesisProducer` diinisialisasi, mengubah `KinesisProducerConfiguration` instance yang digunakan tidak memiliki efek lebih lanjut. `KinesisProducer`saat ini tidak mendukung konfigurasi ulang dinamis.
# Menerapkan de-agregasi konsumen
Dimulai dengan rilis 1.4.0, KCL mendukung de-agregasi otomatis catatan pengguna KPL. Kode aplikasi konsumen yang ditulis dengan versi KCL sebelumnya akan dikompilasi tanpa modifikasi apa pun setelah Anda memperbarui KCL. Namun, jika agregasi KPL digunakan di sisi produsen, ada kehalusan yang melibatkan checkpointing: semua subrecord dalam catatan agregat memiliki nomor urut yang sama, sehingga data tambahan harus disimpan dengan pos pemeriksaan jika Anda perlu membedakan antara subrecord. Data tambahan ini disebut sebagai *nomor urutan*.
**Topics**
+ [Migrasi dari versi KCL sebelumnya](#kinesis-kpl-consumer-deaggregation-migration)
+ [Gunakan ekstensi KCL untuk de-agregasi KPL](#kinesis-kpl-consumer-deaggregation-extensions)
+ [Gunakan GetRecords secara langsung](#kinesis-kpl-consumer-deaggregation-getrecords)
## Migrasi dari versi KCL sebelumnya
Anda tidak diharuskan mengubah panggilan yang ada untuk melakukan checkpointing dengan agregasi. Masih dijamin bahwa Anda dapat mengambil semua catatan yang berhasil disimpan di Kinesis Data Streams. KCL sekarang menyediakan dua operasi pos pemeriksaan baru untuk mendukung kasus penggunaan tertentu, dijelaskan berikut.
Jika kode Anda yang ada ditulis untuk KCL sebelum dukungan KPL, dan operasi pos pemeriksaan Anda dipanggil tanpa argumen, itu setara dengan memeriksa nomor urut catatan pengguna KPL terakhir dalam batch. Jika operasi pos pemeriksaan Anda dipanggil dengan string nomor urut, itu setara dengan pemeriksaan nomor urut yang diberikan dari batch bersama dengan nomor urutan implisit 0 (nol).
Memanggil operasi pos pemeriksaan KCL baru `checkpoint()` tanpa argumen apa pun secara semantik setara dengan pemeriksaan nomor urut `Record` panggilan terakhir dalam batch, bersama dengan nomor urutan implisit 0 (nol).
Memanggil operasi pos pemeriksaan KCL baru `checkpoint(Record record)` secara semantik setara dengan pemeriksaan nomor urut yang diberikan `Record` bersama dengan nomor urutan implisit 0 (nol). Jika `Record` panggilan sebenarnya a`UserRecord`, nomor `UserRecord` urut dan nomor urutan diperiksa.
Memanggil operasi pos pemeriksaan KCL baru `checkpoint(String sequenceNumber, long subSequenceNumber)` secara eksplisit memeriksa nomor urut yang diberikan bersama dengan nomor urutan yang diberikan.
Dalam salah satu kasus ini, setelah pos pemeriksaan disimpan di tabel pos pemeriksaan Amazon DynamoDB, KCL dapat melanjutkan pengambilan catatan dengan benar bahkan ketika aplikasi mogok dan restart. Jika lebih banyak catatan terkandung dalam urutan, pengambilan terjadi dimulai dengan catatan nomor urutan berikutnya dalam catatan dengan nomor urut yang terakhir diperiksa. Jika pos pemeriksaan terbaru menyertakan nomor urutan terakhir dari catatan nomor urut sebelumnya, pengambilan terjadi dimulai dengan catatan dengan nomor urut berikutnya.
Bagian selanjutnya membahas rincian urutan dan urutan checkpointing untuk konsumen yang harus menghindari melewatkan dan duplikasi catatan. Jika melewatkan (atau duplikasi) catatan saat menghentikan dan memulai ulang pemrosesan catatan konsumen Anda tidak penting, Anda dapat menjalankan kode yang ada tanpa modifikasi.
## Gunakan ekstensi KCL untuk de-agregasi KPL
De-agregasi KPL dapat melibatkan checkpointing berikutnya. Untuk memfasilitasi penggunaan checkpointing berikutnya, sebuah `UserRecord` kelas telah ditambahkan ke KCL:
```
public class UserRecord extends Record {
public long getSubSequenceNumber() {
/* ... */
}
@Override
public int hashCode() {
/* contract-satisfying implementation */
}
@Override
public boolean equals(Object obj) {
/* contract-satisfying implementation */
}
}
```
Kelas ini sekarang digunakan sebagai pengganti`Record`. Ini tidak merusak kode yang ada karena merupakan subkelas dari. `Record` `UserRecord`Kelas mewakili subrecord aktual dan standar, catatan non-agregat. Catatan non-agregat dapat dianggap sebagai catatan agregat dengan tepat satu subrecord.
Selain itu, dua operasi baru ditambahkan ke`IRecordProcessorCheckpointer`:
```
public void checkpoint(Record record);
public void checkpoint(String sequenceNumber, long subSequenceNumber);
```
Untuk mulai menggunakan checkpointing nomor urutan, Anda dapat melakukan konversi berikut. Ubah kode formulir berikut:
```
checkpointer.checkpoint(record.getSequenceNumber());
```
Kode formulir baru:
```
checkpointer.checkpoint(record);
```
Kami menyarankan Anda menggunakan `checkpoint(Record record)` formulir untuk checkpointing berikutnya. Namun, jika Anda sudah menyimpan `sequenceNumbers` dalam string untuk digunakan untuk checkpointing, Anda sekarang juga harus menyimpan`subSequenceNumber`, seperti yang ditunjukkan pada contoh berikut:
```
String sequenceNumber = record.getSequenceNumber();
long subSequenceNumber = ((UserRecord) record).getSubSequenceNumber(); // ... do other processing
checkpointer.checkpoint(sequenceNumber, subSequenceNumber);
```
Pemeran dari `Record` untuk `UserRecord` selalu berhasil karena implementasinya selalu menggunakan`UserRecord`. Kecuali ada kebutuhan untuk melakukan aritmatika pada nomor urut, pendekatan ini tidak disarankan.
Saat memproses catatan pengguna KPL, KCL menulis nomor urutan ke Amazon DynamoDB sebagai bidang tambahan untuk setiap baris. Versi KCL sebelumnya digunakan `AFTER_SEQUENCE_NUMBER` untuk mengambil catatan saat melanjutkan pos pemeriksaan. KCL saat ini dengan dukungan KPL menggunakan `AT_SEQUENCE_NUMBER` sebagai gantinya. Ketika catatan pada nomor urut yang diperiksa diambil, nomor urutan yang diperiksa diperiksa, dan subrecord dijatuhkan sebagaimana mestinya (yang mungkin semuanya, jika subrecord terakhir adalah yang diperiksa). Sekali lagi, catatan non-agregat dapat dianggap sebagai catatan agregat dengan satu subrecord, sehingga algoritma yang sama berfungsi untuk catatan agregat dan non-agregat.
## Gunakan GetRecords secara langsung
Anda juga dapat memilih untuk tidak menggunakan KCL tetapi menjalankan operasi API `GetRecords` secara langsung untuk mengambil catatan Kinesis Data Streams. Untuk membongkar catatan yang diambil ini ke dalam catatan pengguna KPL asli Anda, panggil salah satu operasi statis berikut di: `UserRecord.java`
```
public static List deaggregate(List records)
public static List deaggregate(List records, BigInteger startingHashKey, BigInteger endingHashKey)
```
Operasi pertama menggunakan nilai default `0` (nol) untuk `startingHashKey` dan nilai default `2^128 -1` untuk`endingHashKey`.
Masing-masing operasi ini melakukan de-agregasi daftar catatan Kinesis Data Streams yang diberikan ke dalam daftar catatan pengguna KPL. Setiap catatan pengguna KPL yang kunci hash eksplisit atau kunci partisi berada di luar jangkauan `startingHashKey` (inklusif) dan `endingHashKey` (inklusif) dibuang dari daftar catatan yang dikembalikan.
# Gunakan KPL dengan Amazon Data Firehose
Jika Anda menggunakan Kinesis Producer Library (KPL) untuk menulis data ke aliran data Kinesis, Anda dapat menggunakan agregasi untuk menggabungkan catatan yang Anda tulis ke aliran data Kinesis tersebut. Jika Anda kemudian menggunakan aliran data tersebut sebagai sumber untuk aliran pengiriman Firehose Anda, Firehose menghapus agregasi catatan sebelum mengirimkannya ke tujuan. Jika Anda mengonfigurasi aliran pengiriman untuk mengubah data, Firehose menghapus agregasi catatan sebelum mengirimkannya. AWS Lambda Untuk informasi selengkapnya, lihat [Menulis ke Amazon Firehose Menggunakan Kinesis Data Streams](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-kinesis-streams.html).
# Gunakan KPL dengan AWS Glue Schema Registry
Anda dapat mengintegrasikan aliran data Kinesis Anda dengan Registri Skema. AWS Glue Registri AWS Glue Skema memungkinkan Anda untuk menemukan, mengontrol, dan mengembangkan skema secara terpusat, sambil memastikan data yang dihasilkan terus divalidasi oleh skema terdaftar. Sebuah skema mendefinisikan struktur dan format catatan data. Sebuah skema adalah sebuah spesifikasi berversi untuk publikasi data yang handal, konsumsi, atau penyimpanan. Registri AWS Glue Skema memungkinkan Anda untuk meningkatkan kualitas end-to-end data dan tata kelola data dalam aplikasi streaming Anda. Untuk informasi selengkapnya, lihat [AWS Glue Schema Registry](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html). Salah satu cara untuk mengatur integrasi ini adalah melalui perpustakaan KPL dan Kinesis Client Library (KCL) di Jawa.
**penting**
Saat ini, Kinesis Data AWS Glue Streams dan integrasi registri skema hanya didukung untuk aliran data Kinesis yang menggunakan produsen KPL yang diimplementasikan di Jawa. Support multi-bahasa tidak tersedia.
Untuk petunjuk terperinci tentang cara mengatur integrasi Kinesis Data Streams dengan Schema Registry menggunakan KPL, lihat [bagian “Berinteraksi dengan Data Menggunakan Pustaka” di Kasus KPL/KCL Penggunaan: Mengintegrasikan Amazon Kinesis Data Streams](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) dengan Glue Schema Registry. AWS
# Konfigurasikan konfigurasi proxy KPL
Untuk aplikasi yang tidak dapat terhubung langsung ke internet, semua klien AWS SDK mendukung penggunaan proxy HTTP atau HTTPS. Dalam lingkungan perusahaan yang khas, semua lalu lintas jaringan keluar harus melalui server proxy. Jika aplikasi Anda menggunakan Kinesis Producer Library (KPL) untuk mengumpulkan dan mengirim data ke AWS lingkungan yang menggunakan server proxy, aplikasi Anda akan memerlukan konfigurasi proxy KPL. KPL adalah perpustakaan tingkat tinggi yang dibangun di atas AWS Kinesis SDK. Ini dibagi menjadi proses asli dan pembungkus. Proses asli melakukan semua pekerjaan pemrosesan dan pengiriman catatan, sementara pembungkus mengelola proses asli dan berkomunikasi dengannya. Untuk informasi selengkapnya, lihat [Menerapkan Produsen yang Efisien dan Andal dengan Perpustakaan Produsen Amazon Kinesis](https://aws.amazon.com/blogs/big-data/implementing-efficient-and-reliable-producers-with-the-amazon-kinesis-producer-library/).
Pembungkus ditulis dalam Java dan proses asli ditulis dalam C \$1\$1 dengan menggunakan Kinesis SDK. KPL versi 0.14.7 dan yang lebih tinggi sekarang mendukung konfigurasi proxy di pembungkus Java yang dapat meneruskan semua konfigurasi proxy ke proses asli. Untuk informasi lebih lanjut, lihat [https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v0.14.7](https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v0.14.7).
Anda dapat menggunakan kode berikut untuk menambahkan konfigurasi proxy ke aplikasi KPL Anda.
```
KinesisProducerConfiguration configuration = new KinesisProducerConfiguration();
// Next 4 lines used to configure proxy
configuration.setProxyHost("10.0.0.0"); // required
configuration.setProxyPort(3128); // default port is set to 443
configuration.setProxyUserName("username"); // no default
configuration.setProxyPassword("password"); // no default
KinesisProducer kinesisProducer = new KinesisProducer(configuration);
```
# Kebijakan siklus hidup versi KPL
Topik ini menguraikan kebijakan siklus hidup versi untuk Amazon Kinesis Producer Library (KPL). AWS secara teratur menyediakan rilis baru untuk versi KPL untuk mendukung fitur dan penyempurnaan baru, perbaikan bug, patch keamanan, dan pembaruan ketergantungan. Kami menyarankan Anda tetap up-to-date menggunakan versi KPL untuk mengikuti fitur-fitur terbaru, pembaruan keamanan, dan dependensi yang mendasarinya. Kami **tidak** menyarankan untuk terus menggunakan versi KPL yang tidak didukung.
Siklus hidup untuk versi KPL utama terdiri dari tiga fase berikut:
+ **Ketersediaan umum (GA)** — Selama fase ini, versi utama didukung sepenuhnya. AWS menyediakan rilis versi minor dan patch reguler yang mencakup dukungan untuk fitur baru atau pembaruan API untuk Kinesis Data Streams, serta perbaikan bug dan keamanan.
+ **Mode pemeliharaan** — AWS membatasi rilis versi patch untuk mengatasi perbaikan bug kritis dan masalah keamanan saja. Versi utama tidak akan menerima pembaruan untuk fitur baru atau APIs Kinesis Data Streams.
+ **E nd-of-support** — Versi utama tidak akan lagi menerima pembaruan atau rilis. Rilis yang diterbitkan sebelumnya akan terus tersedia melalui manajer paket publik dan kode akan tetap aktif GitHub. Penggunaan versi yang telah end-of-support dicapai dilakukan atas kebijaksanaan pengguna. Kami menyarankan Anda meningkatkan ke versi utama terbaru.
| Versi utama | Fase saat ini | Tanggal rilis | Tanggal mode pemeliharaan | End-of-support tanggal |
| --- | --- | --- | --- | --- |
| KPL 0.x | Mode pemeliharaan | 2015-06-02 | 2025-04-17 | 2026-01-30 |
| KPL 1.x | Ketersediaan umum | 2024-12-15 | -- | -- |
# Kembangkan produsen menggunakan API Amazon Kinesis Data Streams dengan AWS SDK untuk Java
Anda dapat mengembangkan produsen menggunakan Amazon Kinesis Data Streams API AWS dengan SDK for Java. Jika Anda baru mengenal Kinesis Data Streams, mulailah dengan menjadi akrab dengan konsep dan terminologi [Apa itu Amazon Kinesis Data Streams?](introduction.md) yang disajikan dalam dan. [Gunakan AWS CLI untuk menjalankan operasi Amazon Kinesis Data Streams](getting-started.md)
Contoh-contoh ini membahas [Kinesis Data](https://docs.aws.amazon.com/kinesis/latest/APIReference/) Streams API dan [AWS menggunakan SDK for Java untuk](https://aws.amazon.com/sdk-for-java/) menambahkan (menempatkan) data ke stream. Namun, untuk sebagian besar kasus penggunaan, Anda harus memilih pustaka KPL Kinesis Data Streams. Untuk informasi selengkapnya, lihat [Kembangkan produsen menggunakan Amazon Kinesis Producer Library (KPL)](developing-producers-with-kpl.md).
Contoh kode Java dalam chapter ini menunjukkan bagaimana melakukan operasi API Kinesis Data Streams dasar, dan dibagi secara logis berdasarkan jenis operasi. Contoh-contoh ini tidak mewakili kode siap produksi, karena contoh ini tidak memeriksa semua kemungkinan pengecualian, atau memperhitungkan semua kemungkinan pertimbangan keamanan atau performa. Selain itu, Anda dapat memanggil [Kinesis Data](https://docs.aws.amazon.com/kinesis/latest/APIReference/) Streams API menggunakan bahasa pemrograman lainnya. Untuk informasi selengkapnya tentang semua yang tersedia AWS SDKs, lihat [Mulai Mengembangkan dengan Amazon Web Services](https://aws.amazon.com/developers/getting-started/).
Setiap tugas memiliki prasyarat; misalnya, Anda tidak dapat menambahkan data ke aliran sampai Anda membuat aliran, yang mengharuskan Anda membuat klien. Untuk informasi selengkapnya, lihat [Membuat dan mengelola aliran data Kinesis](working-with-streams.md).
**Topics**
+ [Tambahkan data ke aliran](#kinesis-using-sdk-java-add-data-to-stream)
+ [Berinteraksi dengan data menggunakan AWS Glue Schema Registry](kinesis-integration-glue-schema-registry.md)
## Tambahkan data ke aliran
Setelah aliran dibuat, Anda dapat menambahkan data ke dalamnya dalam bentuk catatan. Catatan adalah struktur data yang berisi data yang akan diproses dalam bentuk gumpalan data. Setelah Anda menyimpan data dalam catatan, Kinesis Data Streams tidak memeriksa, menafsirkan, atau mengubah data dengan cara apa pun. Setiap catatan juga memiliki nomor urut dan kunci partisi terkait.
Ada dua operasi berbeda di Kinesis Data Streams API yang menambahkan data ke aliran, dan. [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) `PutRecords`Operasi mengirimkan beberapa catatan ke aliran Anda per permintaan HTTP, dan `PutRecord` operasi tunggal mengirimkan catatan ke aliran Anda satu per satu (permintaan HTTP terpisah diperlukan untuk setiap rekaman). Anda sebaiknya lebih suka menggunakan `PutRecords` untuk sebagian besar aplikasi karena akan mencapai throughput yang lebih tinggi per produsen data. Untuk informasi selengkapnya tentang masing-masing operasi ini, lihat subbagian terpisah di bawah ini.
**Topics**
+ [Tambahkan beberapa catatan dengan PutRecords](#kinesis-using-sdk-java-putrecords)
+ [Tambahkan satu catatan dengan PutRecord](#kinesis-using-sdk-java-putrecord)
Selalu ingat bahwa, karena aplikasi sumber Anda menambahkan data ke aliran menggunakan Kinesis Data Streams API, kemungkinan besar ada satu atau lebih aplikasi konsumen yang secara bersamaan memproses data dari aliran. Untuk informasi tentang cara konsumen mendapatkan data menggunakan Kinesis Data Streams API, lihat. [Dapatkan data dari aliran](developing-consumers-with-sdk.md#kinesis-using-sdk-java-get-data)
**penting**
[Ubah periode retensi data](kinesis-extended-retention.md)
### Tambahkan beberapa catatan dengan PutRecords
[https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)Operasi mengirimkan beberapa catatan ke Kinesis Data Streams dalam satu permintaan. Dengan menggunakan`PutRecords`, produsen dapat mencapai throughput yang lebih tinggi saat mengirim data ke aliran data Kinesis mereka. Setiap `PutRecords` permintaan dapat mendukung hingga 500 catatan. Setiap catatan dalam permintaan dapat sebesar 1 MB, hingga batas 5 MB untuk seluruh permintaan, termasuk kunci partisi. Seperti `PutRecord` operasi tunggal yang dijelaskan di bawah ini, `PutRecords` menggunakan nomor urut dan kunci partisi. Namun, `PutRecord` parameter `SequenceNumberForOrdering` tidak termasuk dalam `PutRecords` panggilan. `PutRecords`Operasi mencoba untuk memproses semua catatan dalam urutan alami permintaan.
Setiap catatan data memiliki nomor urut yang unik. Nomor urutan ditetapkan oleh Kinesis Data Streams setelah `client.putRecords` Anda memanggil untuk menambahkan catatan data ke aliran. Nomor urutan untuk kunci partisi yang sama umumnya meningkat dari waktu ke waktu; semakin lama periode waktu antara `PutRecords` permintaan, semakin besar nomor urut menjadi.
**catatan**
Nomor urutan tidak dapat digunakan sebagai indeks untuk kumpulan data dalam aliran yang sama. Untuk memisahkan kumpulan data secara logis, gunakan kunci partisi atau buat aliran terpisah untuk setiap kumpulan data.
`PutRecords`Permintaan dapat menyertakan catatan dengan kunci partisi yang berbeda. Ruang lingkup permintaan adalah aliran; setiap permintaan dapat mencakup kombinasi kunci partisi dan catatan hingga batas permintaan. Permintaan yang dibuat dengan banyak kunci partisi yang berbeda untuk streaming dengan banyak pecahan yang berbeda umumnya lebih cepat daripada permintaan dengan sejumlah kecil kunci partisi ke sejumlah kecil pecahan. Jumlah kunci partisi harus jauh lebih besar daripada jumlah pecahan untuk mengurangi latensi dan memaksimalkan throughput.
#### PutRecords contoh
Kode berikut membuat 100 catatan data dengan kunci partisi berurutan dan menempatkannya dalam aliran yang disebut`DataStream`.
```
AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
clientBuilder.setRegion(regionName);
clientBuilder.setCredentials(credentialsProvider);
clientBuilder.setClientConfiguration(config);
AmazonKinesis kinesisClient = clientBuilder.build();
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(streamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int i = 0; i < 100; i++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(i).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", i));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = kinesisClient.putRecords(putRecordsRequest);
System.out.println("Put Result" + putRecordsResult);
```
`PutRecords`Responsnya mencakup serangkaian respons`Records`. Setiap catatan dalam larik respons berkorelasi langsung dengan catatan dalam array permintaan menggunakan urutan alami, dari atas ke bawah permintaan dan respons. `Records`Array respons selalu menyertakan jumlah catatan yang sama dengan array permintaan.
#### Menangani kegagalan saat menggunakan PutRecords
Secara default, kegagalan catatan individu dalam permintaan tidak menghentikan pemrosesan catatan berikutnya dalam `PutRecords` permintaan. Ini berarti bahwa `Records` array respons mencakup catatan yang berhasil dan tidak berhasil diproses. Anda harus mendeteksi catatan yang tidak berhasil diproses dan memasukkannya ke dalam panggilan berikutnya.
Catatan yang berhasil termasuk `SequenceNumber` dan `ShardID` nilai, dan catatan yang gagal termasuk `ErrorCode` dan `ErrorMessage` nilai. `ErrorCode`Parameter mencerminkan jenis kesalahan dan dapat menjadi salah satu dari nilai berikut: `ProvisionedThroughputExceededException` atau`InternalFailure`. `ErrorMessage`memberikan informasi lebih rinci tentang `ProvisionedThroughputExceededException` pengecualian termasuk ID akun, nama aliran, dan ID pecahan dari catatan yang dibatasi. Contoh di bawah ini memiliki tiga catatan dalam satu `PutRecords` permintaan. Catatan kedua gagal dan tercermin dalam respons.
**Example PutRecords Permintaan Sintaks**
```
{
"Records": [
{
"Data": "XzxkYXRhPl8w",
"PartitionKey": "partitionKey1"
},
{
"Data": "AbceddeRFfg12asd",
"PartitionKey": "partitionKey1"
},
{
"Data": "KFpcd98*7nd1",
"PartitionKey": "partitionKey3"
}
],
"StreamName": "myStream"
}
```
**Example PutRecords Sintaks Respon**
```
{
"FailedRecordCount”: 1,
"Records": [
{
"SequenceNumber": "21269319989900637946712965403778482371",
"ShardId": "shardId-000000000001"
},
{
“ErrorCode":”ProvisionedThroughputExceededException”,
“ErrorMessage": "Rate exceeded for shard shardId-000000000001 in stream exampleStreamName under account 111111111111."
},
{
"SequenceNumber": "21269319989999637946712965403778482985",
"ShardId": "shardId-000000000002"
}
]
}
```
Catatan yang tidak berhasil diproses dapat dimasukkan dalam `PutRecords` permintaan berikutnya. Pertama, periksa parameter `FailedRecordCount` dalam `putRecordsResult` untuk mengonfirmasi apakah ada catatan kegagalan dalam permintaan. Jika demikian, masing-masing `putRecordsEntry` yang memiliki `ErrorCode` yang tidak `null` harus ditambahkan ke permintaan berikutnya. Untuk contoh jenis handler ini, lihat kode berikut.
**Example PutRecords penangan kegagalan**
```
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(myStreamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int j = 0; j < 100; j++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(j).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", j));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
while (putRecordsResult.getFailedRecordCount() > 0) {
final List failedRecordsList = new ArrayList<>();
final List putRecordsResultEntryList = putRecordsResult.getRecords();
for (int i = 0; i < putRecordsResultEntryList.size(); i++) {
final PutRecordsRequestEntry putRecordRequestEntry = putRecordsRequestEntryList.get(i);
final PutRecordsResultEntry putRecordsResultEntry = putRecordsResultEntryList.get(i);
if (putRecordsResultEntry.getErrorCode() != null) {
failedRecordsList.add(putRecordRequestEntry);
}
}
putRecordsRequestEntryList = failedRecordsList;
putRecordsRequest.setRecords(putRecordsRequestEntryList);
putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
}
```
### Tambahkan satu catatan dengan PutRecord
Setiap panggilan untuk [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html)beroperasi pada satu catatan. Lebih suka `PutRecords` operasi yang dijelaskan [Tambahkan beberapa catatan dengan PutRecords](#kinesis-using-sdk-java-putrecords) kecuali aplikasi Anda secara khusus perlu selalu mengirim catatan tunggal per permintaan, atau alasan lain tidak `PutRecords` dapat digunakan.
Setiap catatan data memiliki nomor urut yang unik. Nomor urutan ditetapkan oleh Kinesis Data Streams setelah `client.putRecord` Anda memanggil untuk menambahkan catatan data ke aliran. Nomor urutan untuk kunci partisi yang sama umumnya meningkat dari waktu ke waktu; semakin lama periode waktu antara `PutRecord` permintaan, semakin besar nomor urut menjadi.
Ketika put terjadi secara berurutan, nomor urut yang dikembalikan tidak dijamin akan meningkat karena operasi put pada dasarnya muncul bersamaan dengan Kinesis Data Streams. Untuk menjamin peningkatan nomor urut secara ketat untuk kunci partisi yang sama, gunakan `SequenceNumberForOrdering` parameter, seperti yang ditunjukkan pada sampel [PutRecord contoh](#kinesis-using-sdk-java-putrecord-example) kode.
Apakah Anda menggunakan atau tidak`SequenceNumberForOrdering`, catatan yang diterima Kinesis Data Streams `GetRecords` melalui panggilan diurutkan secara ketat berdasarkan nomor urut.
**catatan**
Nomor urutan tidak dapat digunakan sebagai indeks untuk kumpulan data dalam aliran yang sama. Untuk memisahkan kumpulan data secara logis, gunakan kunci partisi atau buat aliran terpisah untuk setiap kumpulan data.
Kunci partisi digunakan untuk mengelompokkan data dalam aliran. Catatan data ditugaskan ke pecahan dalam aliran berdasarkan kunci partisi. Secara khusus, Kinesis Data Streams menggunakan kunci partisi sebagai input ke fungsi hash yang memetakan kunci partisi (dan data terkait) ke pecahan tertentu.
Sebagai hasil dari mekanisme hashing ini, semua data mencatat dengan kunci partisi yang sama memetakan ke pecahan yang sama dalam aliran. Namun, jika jumlah kunci partisi melebihi jumlah pecahan, beberapa pecahan harus berisi catatan dengan kunci partisi yang berbeda. Dari sudut pandang desain, untuk memastikan bahwa semua pecahan Anda digunakan dengan baik, jumlah pecahan (ditentukan oleh `setShardCount` metode`CreateStreamRequest`) harus jauh lebih sedikit daripada jumlah kunci partisi unik, dan jumlah data yang mengalir ke kunci partisi tunggal harus jauh lebih kecil dari kapasitas pecahan.
#### PutRecord contoh
Kode berikut membuat sepuluh catatan data, didistribusikan di dua kunci partisi, dan menempatkan mereka dalam aliran yang disebut`myStreamName`.
```
for (int j = 0; j < 10; j++)
{
PutRecordRequest putRecordRequest = new PutRecordRequest();
putRecordRequest.setStreamName( myStreamName );
putRecordRequest.setData(ByteBuffer.wrap( String.format( "testData-%d", j ).getBytes() ));
putRecordRequest.setPartitionKey( String.format( "partitionKey-%d", j/5 ));
putRecordRequest.setSequenceNumberForOrdering( sequenceNumberOfPreviousRecord );
PutRecordResult putRecordResult = client.putRecord( putRecordRequest );
sequenceNumberOfPreviousRecord = putRecordResult.getSequenceNumber();
}
```
Contoh kode sebelumnya digunakan `setSequenceNumberForOrdering` untuk menjamin peningkatan pemesanan secara ketat dalam setiap kunci partisi. Untuk menggunakan parameter ini secara efektif, atur catatan saat ini (record *n*) ke nomor urut catatan sebelumnya (record *n-1*). `SequenceNumberForOrdering` Untuk mendapatkan nomor urut rekaman yang telah ditambahkan ke aliran, panggil `getSequenceNumber` hasil dari`putRecord`.
`SequenceNumberForOrdering`Parameter memastikan secara ketat meningkatkan nomor urut untuk kunci partisi yang sama. `SequenceNumberForOrdering`tidak menyediakan urutan catatan di beberapa kunci partisi.
# Berinteraksi dengan data menggunakan AWS Glue Schema Registry
Anda dapat mengintegrasikan aliran data Kinesis Anda dengan Registri Skema. AWS Glue Registri AWS Glue Skema memungkinkan Anda untuk menemukan, mengontrol, dan mengembangkan skema secara terpusat, sambil memastikan data yang dihasilkan terus divalidasi oleh skema terdaftar. Sebuah skema mendefinisikan struktur dan format catatan data. Sebuah skema adalah sebuah spesifikasi berversi untuk publikasi data yang handal, konsumsi, atau penyimpanan. Registri AWS Glue Skema memungkinkan Anda meningkatkan kualitas end-to-end data dan tata kelola data dalam aplikasi streaming Anda. Untuk informasi selengkapnya, lihat [AWS Glue Schema Registry](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html). Salah satu cara untuk mengatur integrasi ini adalah melalui `PutRecords` dan `PutRecord` Kinesis APIs Data Streams yang tersedia AWS di Java SDK.
Untuk petunjuk terperinci tentang cara mengatur integrasi Kinesis Data Streams dengan registri skema menggunakan dan Kinesis Data Streams, lihat bagian “Berinteraksi dengan Data PutRecords Menggunakan PutRecord Aliran Data APIs Kinesis” di Kasus [Penggunaan: Mengintegrasikan](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) Amazon Kinesis Data APIs Streams dengan Registri Skema Glue. AWS
# Menulis ke Amazon Kinesis Data Streams menggunakan Agen Kinesis
Kinesis Agent adalah aplikasi perangkat lunak Java yang berdiri sendiri yang menawarkan cara mudah untuk mengumpulkan dan mengirim data ke Kinesis Data Streams. Agen terus memantau serangkaian file dan mengirimkan data baru ke aliran Anda. Agen menangani rotasi file, checkpointing, dan coba lagi pada kegagalan. Agen memberikan semua data Anda dengan cara yang andal, tepat waktu, dan sederhana. Ini juga memancarkan CloudWatch metrik Amazon untuk membantu Anda memantau dan memecahkan masalah proses streaming dengan lebih baik.
Secara default, catatan diurai dari setiap file berdasarkan karakter baris baru (`'\n'`). Namun, agen juga dapat dikonfigurasi untuk mengurai catatan multi-baris (lihat [Tentukan pengaturan konfigurasi agen](#agent-config-settings)).
Anda dapat menginstal agen di lingkungan server berbasis Linux seperti server web, server log, dan server basis data. Setelah menginstal agen, konfigurasikan dengan menentukan file yang akan dipantau dan aliran untuk data. Setelah agen dikonfigurasi, agen akan mengumpulkan data dari file dengan andal dan mengirimkannya ke aliran dengan andal.
**Topics**
+ [Lengkapi prasyarat untuk Agen Kinesis](#prereqs)
+ [Unduh dan instal agen](#download-install)
+ [Konfigurasikan dan mulai agen](#config-start)
+ [Tentukan pengaturan konfigurasi agen](#agent-config-settings)
+ [Pantau beberapa direktori file dan tulis ke beberapa aliran](#sim-writes)
+ [Gunakan agen untuk melakukan pra-proses data](#pre-processing)
+ [Gunakan perintah CLI agen](#cli-commands)
+ [Pertanyaan yang Sering Diajukan](#agent-faq)
## Lengkapi prasyarat untuk Agen Kinesis
+ Sistem operasi Anda harus Amazon Linux AMI dengan versi 2015.09 atau yang lebih baru, atau Red Hat Enterprise Linux versi 7 atau yang lebih baru.
+ Jika Anda menggunakan Amazon EC2 untuk menjalankan agen Anda, luncurkan instans EC2 Anda.
+ Kelola AWS kredensyal Anda menggunakan salah satu metode berikut:
+ Tentukan IAM role ketika Anda meluncurkan instans EC2 Anda.
+ Tentukan AWS kredensil saat Anda mengonfigurasi agen (lihat [awsAccessKeyId](#awsAccessKeyId) dan [awsSecretAccessKunci](#awsSecretAccessKey)).
+ Edit `/etc/sysconfig/aws-kinesis-agent` untuk menentukan wilayah dan kunci AWS akses Anda.
+ [[Jika instans EC2 Anda berada di AWS akun yang berbeda, buat peran IAM untuk menyediakan akses ke layanan Kinesis Data Streams, dan tentukan peran tersebut saat Anda mengonfigurasi agen (lihat AssumeroLearn dan Id). assumeRoleExternal](#assumeRoleExternalId)](#assumeRoleARN) Gunakan salah satu metode sebelumnya untuk menentukan AWS kredensi pengguna di akun lain yang memiliki izin untuk mengambil peran ini.
+ Peran IAM atau AWS kredenal yang Anda tentukan harus memiliki izin untuk melakukan operasi Kinesis Data [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)Streams agar agen dapat mengirim data ke aliran Anda. Jika Anda mengaktifkan CloudWatch pemantauan untuk agen, izin untuk melakukan CloudWatch [PutMetricData](https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_PutMetricData.html)operasi juga diperlukan. Untuk informasi selengkapnya[Mengontrol akses ke sumber daya Amazon Kinesis Data Streams menggunakan IAM](controlling-access.md), lihat[Pantau kesehatan Agen Aliran Data Kinesis dengan Amazon CloudWatch](agent-health.md), dan [Kontrol CloudWatch Akses](https://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/UsingIAM.html).
## Unduh dan instal agen
Pertama-tama, hubungkan ke instans Anda. Untuk informasi selengkapnya, lihat [Connect to Your Instance](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-connect-to-instance-linux.html) di *Panduan Pengguna Amazon EC2*. Jika Anda mengalami masalah saat menyambung, lihat [Pemecahan Masalah Menyambung ke Instans Anda](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/TroubleshootingInstancesConnecting.html) di Panduan Pengguna *Amazon EC2*.
**Untuk mengatur agen menggunakan Amazon Linux AMI**
Gunakan perintah berikut untuk mengunduh dan menginstal agen:
```
sudo yum install –y aws-kinesis-agent
```
**Untuk mengatur agen menggunakan Red Hat Enterprise Linux**
Gunakan perintah berikut untuk mengunduh dan menginstal agen:
```
sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm
```
**Untuk mengatur agen menggunakan GitHub**
1. Unduh agen dari [amazon-kinesis-agentawlabs/](https://github.com/awslabs/amazon-kinesis-agent).
1. Instal agen dengan menavigasi ke direktori unduhan dan menjalankan perintah berikut:
```
sudo ./setup --install
```
**Untuk mengatur agen dalam wadah Docker**
Agen Kinesis dapat dijalankan dalam wadah juga melalui basis wadah [amazonlinux](https://docs.aws.amazon.com/AmazonECR/latest/userguide/amazon_linux_container_image.html). Gunakan Dockerfile berikut dan kemudian jalankan. `docker build`
```
FROM amazonlinux
RUN yum install -y aws-kinesis-agent which findutils
COPY agent.json /etc/aws-kinesis/agent.json
CMD ["start-aws-kinesis-agent"]
```
## Konfigurasikan dan mulai agen
**Untuk mengonfigurasi dan memulai agen**
1. Buka dan edit file konfigurasi (sebagai pengguna super jika menggunakan izin akses file default): `/etc/aws-kinesis/agent.json`
Dalam file konfigurasi ini, tentukan file (`"filePattern"`) dari mana agen mengumpulkan data, dan nama stream (`"kinesisStream"`) tempat agen mengirim data. Perhatikan bahwa nama file adalah pola, dan agen mengenali rotasi file. Anda dapat memutar file atau membuat file baru satu kali per detik. Agen menggunakan stempel waktu pembuatan file untuk menentukan file mana yang akan dilacak dan diekor ke aliran Anda; membuat file baru atau memutar file lebih sering dari sekali per detik tidak memungkinkan agen untuk membedakan dengan benar di antara mereka.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "yourkinesisstream"
}
]
}
```
1. Mulailah agen secara manual:
```
sudo service aws-kinesis-agent start
```
1. (Opsional) Konfigurasikan agen untuk memulai pada startup sistem:
```
sudo chkconfig aws-kinesis-agent on
```
Agen sekarang berjalan sebagai layanan sistem di latar belakang. Ini terus memantau file yang ditentukan dan mengirim data ke aliran yang ditentukan. Aktivitas agen masuk di `/var/log/aws-kinesis-agent/aws-kinesis-agent.log`.
## Tentukan pengaturan konfigurasi agen
Agen mendukung dua pengaturan konfigurasi wajib, `filePattern` dan`kinesisStream`, ditambah pengaturan konfigurasi opsional untuk fitur tambahan. Anda dapat menentukan konfigurasi wajib dan opsional di`/etc/aws-kinesis/agent.json`.
Setiap kali mengubah file konfigurasi, Anda harus menghentikan dan memulai agen, menggunakan perintah berikut:
```
sudo service aws-kinesis-agent stop
sudo service aws-kinesis-agent start
```
Atau, Anda dapat menggunakan perintah berikut:
```
sudo service aws-kinesis-agent restart
```
Berikut ini adalah pengaturan konfigurasi umum.
| Pengaturan Konfigurasi | Deskripsi |
| --- | --- |
| assumeRoleARN | ARN dari peran yang akan diasumsikan oleh pengguna. Untuk informasi selengkapnya, lihat [Mendelegasikan Akses di Seluruh AWS Akun Menggunakan Peran IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/tutorial_cross-account-with-roles.html) di Panduan Pengguna *IAM*. |
| assumeRoleExternalId | Pengidentifikasi opsional yang menentukan siapa yang dapat mengambil peran tersebut. Untuk informasi selengkapnya, lihat [Cara Menggunakan ID Eksternal](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html) di *Panduan Pengguna IAM*. |
| awsAccessKeyId | AWS ID kunci akses yang mengesampingkan kredensyal default. Pengaturan ini diutamakan daripada semua penyedia kredensial lainnya. |
| awsSecretAccessKey | AWS kunci rahasia yang mengesampingkan kredensi default. Pengaturan ini diutamakan daripada semua penyedia kredensial lainnya. |
| cloudwatch.emitMetrics | Memungkinkan agen untuk memancarkan metrik ke CloudWatch if set (true). Default: betul |
| cloudwatch.endpoint | Titik akhir regional untuk CloudWatch. Default: `monitoring.us-east-1.amazonaws.com` |
| kinesis.endpoint | Titik akhir regional untuk Kinesis Data Streams. Default: `kinesis.us-east-1.amazonaws.com` |
Berikut ini adalah pengaturan konfigurasi aliran.
| Pengaturan Konfigurasi | Deskripsi |
| --- | --- |
| dataProcessingOptions | Daftar opsi pemrosesan diterapkan ke setiap catatan yang diuraikan sebelum dikirim ke aliran. Pilihan pemrosesan dilakukan dalam urutan yang ditentukan. Untuk informasi selengkapnya, lihat [Gunakan agen untuk melakukan pra-proses data](#pre-processing). |
| kinesisStream | [Wajib] Nama aliran. |
| filePattern | [Wajib] Direktori dan pola file yang harus dicocokkan untuk diambil oleh agen. Untuk semua file yang cocok dengan pola ini, izin baca harus diberikan`aws-kinesis-agent-user`. Untuk direktori yang berisi file, izin baca dan eksekusi harus diberikan kepada`aws-kinesis-agent-user`. |
| initialPosition | Posisi awal dari mana file mulai diurai. Nilai yang valid adalah `START_OF_FILE` dan `END_OF_FILE`. Default: `END_OF_FILE` |
| maxBufferAgeMillis | Waktu maksimum, dalam milidetik, di mana agen menyangga data sebelum mengirimnya ke aliran. Rentang nilai: 1.000 hingga 900.000 (1 detik hingga 15 menit) Default: 60.000 (1 menit) |
| maxBufferSizeBytes | Ukuran maksimum, dalam byte, di mana agen buffer data sebelum mengirimnya ke aliran. Rentang nilai: 1 hingga 4.194.304 (4 MB) Default: 4.194.304 (4 MB) |
| maxBufferSizeRecords | Jumlah maksimum catatan yang agen buffer data sebelum mengirimnya ke aliran. Rentang nilai: 1 hingga 500 Default: 500 |
| minTimeBetweenFilePollsMillis | Interval waktu, dalam milidetik, saat agen melakukan polling dan mengurai file yang dipantau untuk data baru. Kisaran nilai: 1 atau lebih Default: 100 |
| multiLineStartPattern | Pola untuk mengidentifikasi awal catatan. Catatan dibuat dari baris yang cocok dengan pola tersebut dan baris berikutnya yang tidak cocok dengan pola tersebut. Nilai-nilai yang benar adalah ekspresi reguler. Secara default, setiap baris baru dalam file log diurai sebagai satu catatan. |
| partitionKeyOption | Metode untuk menghasilkan kunci partisi. Nilai yang valid adalah `RANDOM` (bilangan bulat yang dihasilkan secara acak) dan `DETERMINISTIC` (nilai hash dihitung dari data). Default: `RANDOM` |
| skipHeaderLines | Jumlah baris yang dilewati agen untuk diurai di awal file yang dipantau. Kisaran nilai: 0 atau lebih Default: 0 (nol) |
| truncatedRecordTerminator | String yang digunakan agen untuk memotong rekaman yang diuraikan ketika ukuran rekaman melebihi batas ukuran rekaman Kinesis Data Streams. (1.000 KB) Default: `'\n'` (baris baru) |
## Pantau beberapa direktori file dan tulis ke beberapa aliran
Dengan menentukan beberapa pengaturan konfigurasi aliran, Anda dapat mengonfigurasi agen untuk memantau beberapa direktori file dan mengirim data ke beberapa aliran. Dalam contoh konfigurasi berikut, agen memonitor dua direktori file dan mengirimkan data ke aliran Kinesis dan aliran pengiriman Firehose masing-masing. Perhatikan bahwa Anda dapat menentukan titik akhir yang berbeda untuk Kinesis Data Streams dan Firehose sehingga aliran Kinesis dan aliran pengiriman Firehose tidak perlu berada di wilayah yang sama.
```
{
"cloudwatch.emitMetrics": true,
"kinesis.endpoint": "https://your/kinesis/endpoint",
"firehose.endpoint": "https://your/firehose/endpoint",
"flows": [
{
"filePattern": "/tmp/app1.log*",
"kinesisStream": "yourkinesisstream"
},
{
"filePattern": "/tmp/app2.log*",
"deliveryStream": "yourfirehosedeliverystream"
}
]
}
```
Untuk informasi lebih lanjut tentang penggunaan agen dengan Firehose, lihat [Menulis ke Amazon Data Firehose dengan](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-agents.html) Agen Kinesis.
## Gunakan agen untuk melakukan pra-proses data
Agen dapat melakukan pra-proses rekaman yang diuraikan dari file yang dipantau sebelum mengirimnya ke streaming Anda. Anda dapat mengaktifkan fitur ini dengan menambahkan pengaturan konfigurasi `dataProcessingOptions` ke aliran file Anda. Satu atau lebih opsi pemrosesan dapat ditambahkan dan mereka akan dilakukan dalam urutan yang ditentukan.
Agen mendukung opsi pemrosesan berikut yang tercantum. Karena agen bersifat open-source, Anda dapat mengembangkan dan memperluas opsi pemrosesannya lebih lanjut. Anda dapat mengunduh agen dari [Agen Kinesis](https://github.com/awslabs/amazon-kinesis-agent).Opsi Pemrosesan
`SINGLELINE`
Mengonversi rekaman multi-baris menjadi catatan baris tunggal dengan menghapus karakter baris baru, spasi utama, dan spasi tambahan.
```
{
"optionName": "SINGLELINE"
}
```
`CSVTOJSON`
Mengkonversi rekaman dari pembatas format dipisahkan ke format JSON.
```
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", ... ],
"delimiter": "yourdelimiter"
}
```
`customFieldNames`
[Diperlukan] Nama-nama field yang digunakan sebagai kunci dalam setiap pasangan nilai kunci JSON. Misalnya, jika Anda menentukan`["f1", "f2"]`, catatan “v1, v2" akan dikonversi ke. `{"f1":"v1","f2":"v2"}`
`delimiter`
String yang digunakan sebagai pembatas dalam catatan. Default adalah koma (,).
`LOGTOJSON`
Mengonversi catatan dari format log ke format JSON. Format log yang didukung adalah **Apache Common Log**, **Apache Combined Log**, **Apache Error Log**, dan **RFC3164 Syslog**.
```
{
"optionName": "LOGTOJSON",
"logFormat": "logformat",
"matchPattern": "yourregexpattern",
"customFieldNames": [ "field1", "field2", … ]
}
```
`logFormat`
[Diperlukan] Format entri log. Berikut adalah nilai yang mungkin:
+ `COMMONAPACHELOG` — Format Log Umum Apache. Setiap entri log memiliki pola berikut secara default: "`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes}`".
+ `COMBINEDAPACHELOG` — Format Log Gabungan Apache. Setiap entri log memiliki pola berikut secara default: "`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes} %{referrer} %{agent}`".
+ `APACHEERRORLOG` — Format Log Kesalahan Apache. Setiap entri log memiliki pola berikut secara default: "`[%{timestamp}] [%{module}:%{severity}] [pid %{processid}:tid %{threadid}] [client: %{client}] %{message}`".
+ `SYSLOG`— Format RFC3164 Syslog. Setiap entri log memiliki pola berikut secara default: "`%{timestamp} %{hostname} %{program}[%{processid}]: %{message}`".
`matchPattern`
Pola ekspresi reguler digunakan untuk mengekstrak nilai dari entri log. Pengaturan ini digunakan jika entri log Anda tidak dalam salah satu format log yang telah ditentukan. Jika pengaturan ini digunakan, Anda juga harus menentukan`customFieldNames`.
`customFieldNames`
Nama bidang khusus digunakan sebagai kunci dalam setiap pasangan nilai kunci JSON. Anda dapat menggunakan pengaturan ini untuk menentukan nama bidang untuk nilai-nilai yang diekstraksi dari `matchPattern`, atau menimpa nama bidang default dari format log yang telah ditetapkan sebelumnya.
**Example : Konfigurasi LOGTOJSON**
Berikut adalah salah satu contoh konfigurasi `LOGTOJSON` untuk entri Log Umum Apache yang dikonversi ke format JSON:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
```
Sebelum konversi:
```
64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291
```
Setelah konversi:
```
{"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"}
```
**Example : Konfigurasi LOGTOJSON dengan Bidang Khusus**
Berikut adalah contoh lain konfigurasi `LOGTOJSON`:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]
}
```
Dengan pengaturan konfigurasi ini, entri Log Umum Apache yang sama dari contoh sebelumnya dikonversi ke format JSON sebagai berikut:
```
{"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"}
```
**Example : Mengonversi Entri Log Umum Apache**
Konfigurasi alur berikut mengonversi entri Apache Common Log ke catatan baris tunggal dalam format JSON:
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"dataProcessingOptions": [
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
]
}
]
}
```
**Example : Mengonversi Catatan Multi-Baris**
Konfigurasi aliran berikut mengurai catatan multi-baris yang baris pertamanya dimulai dengan "`[SEQUENCE=`". Setiap catatan pertama kali dikonversi ke catatan baris tunggal. Kemudian, nilai-nilai diekstraksi dari catatan tersebut berdasarkan pembatas tab. Nilai yang diekstraksi dipetakan ke nilai `customFieldNames` yang ditentukan untuk membentuk catatan baris tunggal dalam format JSON.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"multiLineStartPattern": "\\[SEQUENCE=",
"dataProcessingOptions": [
{
"optionName": "SINGLELINE"
},
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", "field3" ],
"delimiter": "\\t"
}
]
}
]
}
```
**Example : Konfigurasi LOGTOJSON dengan Pola Pencocokan**
Berikut adalah salah satu contoh konfigurasi `LOGTOJSON` untuk entri Log Umum Apache yang dikonversi ke format JSON, dengan bidang terakhir (byte) dihilangkan:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})",
"customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"]
}
```
Sebelum konversi:
```
123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200
```
Setelah konversi:
```
{"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"}
```
## Gunakan perintah CLI agen
Secara otomatis memulai agen pada startup sistem:
```
sudo chkconfig aws-kinesis-agent on
```
Memeriksa status agen:
```
sudo service aws-kinesis-agent status
```
Menghentikan agen:
```
sudo service aws-kinesis-agent stop
```
Membaca file log agen dari lokasi ini:
```
/var/log/aws-kinesis-agent/aws-kinesis-agent.log
```
Menghapus instalasi agen:
```
sudo yum remove aws-kinesis-agent
```
## Pertanyaan yang Sering Diajukan
### Apakah ada Agen Kinesis untuk Windows?
[Kinesis Agent untuk Windows](https://docs.aws.amazon.com/kinesis-agent-windows/latest/userguide/what-is-kinesis-agent-windows.html) adalah perangkat lunak yang berbeda dari Kinesis Agent untuk platform Linux.
### Mengapa Agen Kinesis melambat meningkat? and/or `RecordSendErrors`
Ini biasanya karena pelambatan dari Kinesis. Periksa `WriteProvisionedThroughputExceeded` metrik untuk Kinesis Data Streams `ThrottledRecords` atau metrik untuk Firehose Delivery Streams. Setiap peningkatan dari 0 dalam metrik ini menunjukkan bahwa batas aliran perlu ditingkatkan. Untuk informasi selengkapnya, lihat [batas Kinesis Data Stream](https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html) dan Amazon [Firehose](https://docs.aws.amazon.com/firehose/latest/dev/limits.html) Delivery Streams.
Setelah Anda mengesampingkan pembatasan, lihat apakah Agen Kinesis dikonfigurasi untuk mengekor sejumlah besar file kecil. Ada penundaan ketika Agen Kinesis mengekor file baru, jadi Agen Kinesis harus membuntuti sejumlah kecil file yang lebih besar. Coba konsolidasikan file log Anda ke file yang lebih besar.
### Mengapa saya mendapatkan `java.lang.OutOfMemoryError` pengecualian?
Agen Kinesis tidak memiliki cukup memori untuk menangani beban kerjanya saat ini. Cobalah meningkatkan `JAVA_START_HEAP` dan `JAVA_MAX_HEAP` masuk `/usr/bin/start-aws-kinesis-agent` dan memulai kembali agen.
### Mengapa saya mendapatkan `IllegalStateException : connection pool shut down` pengecualian?
Agen Kinesis tidak memiliki koneksi yang cukup untuk menangani beban kerjanya saat ini. Coba tingkatkan `maxConnections` dan `maxSendingThreads` dalam pengaturan konfigurasi agen umum Anda di`/etc/aws-kinesis/agent.json`. Nilai default untuk bidang ini adalah 12 kali prosesor runtime yang tersedia. Lihat [AgentConfiguration.java](https://github.com/awslabs/amazon-kinesis-agent/blob/master/src/com/amazon/kinesis/streaming/agent/config/AgentConfiguration.java) untuk mengetahui selengkapnya tentang pengaturan konfigurasi agen lanjutan.
### Bagaimana saya bisa men-debug masalah lain dengan Agen Kinesis?
`DEBUG`log level dapat diaktifkan di`/etc/aws-kinesis/log4j.xml`.
### Bagaimana cara mengonfigurasi Agen Kinesis?
Semakin kecil`maxBufferSizeBytes`, semakin sering Agen Kinesis akan mengirim data. Ini bisa bagus karena mengurangi waktu pengiriman catatan, tetapi juga meningkatkan permintaan per detik ke Kinesis.
### Mengapa Agen Kinesis mengirimkan catatan duplikat?
Ini terjadi karena kesalahan konfigurasi dalam file tailing. Pastikan masing-masing hanya `fileFlow’s filePattern` cocok dengan satu file. Ini juga dapat terjadi jika `logrotate` mode yang digunakan dalam `copytruncate` mode. Coba ubah mode ke mode default atau buat untuk menghindari duplikasi. Untuk informasi selengkapnya tentang penanganan rekaman duplikat, lihat [Menangani Rekaman Duplikat.](https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-duplicates.html)
# Menulis ke Kinesis Data AWS Streams menggunakan layanan lain
AWS Layanan berikut dapat diintegrasikan langsung dengan Amazon Kinesis Data Streams untuk menulis data ke aliran data Kinesis. Tinjau informasi untuk setiap layanan yang Anda minati dan lihat referensi yang disediakan.
**Topics**
+ [Menulis ke Kinesis Data Streams menggunakan AWS Amplify](using-other-services-amplify.md)
+ [Menulis ke Kinesis Data Streams menggunakan Amazon Aurora](using-other-services-aurora.md)
+ [Menulis ke Kinesis Data Streams menggunakan Amazon CloudFront](using-other-services-CloudFront.md)
+ [Menulis ke Kinesis Data Streams CloudWatch menggunakan Amazon Logs](using-other-services-cw-logs.md)
+ [Menulis ke Kinesis Data Streams menggunakan Amazon Connect](using-other-services-connect.md)
+ [Menulis ke Kinesis Data Streams menggunakan AWS Database Migration Service](using-other-services-migration.md)
+ [Menulis ke Kinesis Data Streams menggunakan Amazon DynamoDB](using-other-services-ddb.md)
+ [Menulis ke Kinesis Data Streams menggunakan Amazon EventBridge](using-other-services-eventbridges.md)
+ [Menulis ke Kinesis Data Streams menggunakan AWS IoT Core](using-other-services-iot-core.md)
+ [Menulis ke Kinesis Data Streams menggunakan Amazon Relational Database Service](using-other-services-rds.md)
+ [Menulis ke Kinesis Data Streams menggunakanAmazon Pinpoint](using-other-services-pinpoint.md)
+ [Menulis ke Kinesis Data Streams menggunakan Amazon Quantum Ledger Database (Amazon QLDB)](using-other-services-quantum-ledger.md)
# Menulis ke Kinesis Data Streams menggunakan AWS Amplify
Anda dapat menggunakan Amazon Kinesis Data Streams untuk mengalirkan data dari aplikasi seluler yang dibuat AWS dengan Amplify untuk pemrosesan waktu nyata. Anda kemudian dapat membuat dasbor real-time, menangkap pengecualian dan menghasilkan peringatan, mendorong rekomendasi, dan membuat keputusan bisnis atau operasional real-time lainnya. Anda juga dapat mengirim data ke layanan lain seperti Amazon Simple Storage Service, Amazon DynamoDB, dan Amazon Redshift.
Untuk informasi selengkapnya, lihat [Menggunakan Amazon Kinesis](https://docs.amplify.aws/react/build-a-backend/more-features/analytics/streaming-data/) di Pusat Pengembang *AWS Amplify*.
# Menulis ke Kinesis Data Streams menggunakan Amazon Aurora
Anda dapat menggunakan Amazon Kinesis Data Streams untuk memantau aktivitas di kluster Amazon Aurora DB Anda. Menggunakan Aliran Aktivitas Database, klaster Aurora DB Anda mendorong aktivitas ke Aliran Data Amazon Kinesis secara real-time. Anda kemudian dapat membangun aplikasi untuk manajemen kepatuhan yang menggunakan aktivitas ini, mengaudit mereka dan menghasilkan peringatan. Anda juga dapat menggunakan Amazon Amazon Firehose untuk menyimpan data.
Untuk informasi selengkapnya, lihat [Aliran Aktivitas Database](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/DBActivityStreams.html) di Panduan Pengembang *Amazon Aurora*.
# Menulis ke Kinesis Data Streams menggunakan Amazon CloudFront
Anda dapat menggunakan Amazon Kinesis Data CloudFront Streams dengan log real-time dan mendapatkan informasi tentang permintaan yang dibuat ke distribusi secara real time. Anda kemudian dapat membangun [konsumen aliran data Kinesis](https://docs.aws.amazon.com/streams/latest/dev/building-consumers.html) Anda sendiri, atau menggunakan Amazon Data Firehose untuk mengirim data log ke Amazon S3, Amazon Redshift, Amazon Service, atau layanan pemrosesan log pihak OpenSearch ketiga.
Untuk informasi selengkapnya, lihat [Log waktu nyata](https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/real-time-logs.html) di *Panduan CloudFront Pengembang Amazon*.
# Menulis ke Kinesis Data Streams CloudWatch menggunakan Amazon Logs
Anda dapat menggunakan CloudWatch langganan untuk mendapatkan akses ke umpan real-time peristiwa log dari Amazon CloudWatch Logs dan mengirimkannya ke aliran data Kinesis untuk diproses, dianalisis, dan dimuat ke sistem lain.
Untuk informasi selengkapnya, lihat [Pemrosesan data log secara real-time dengan langganan](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Subscriptions.html) di *Panduan Pengguna Amazon CloudWatch Logs*.
# Menulis ke Kinesis Data Streams menggunakan Amazon Connect
Anda dapat menggunakan Kinesis Data Streams untuk mengekspor catatan kontak dan peristiwa agen secara real-time dari instans Amazon Connect. Anda juga dapat mengaktifkan streaming data dari Profil Pelanggan Amazon Connect untuk secara otomatis menerima pembaruan ke aliran data Kinesis tentang pembuatan profil baru atau perubahan pada profil yang sudah ada.
Anda kemudian dapat membangun aplikasi konsumen untuk memproses dan menganalisis data secara real time. Misalnya, menggunakan catatan kontak dan data profil pelanggan, Anda dapat menyimpan data sistem sumber Anda, seperti CRMs dan alat otomatisasi pemasaran, up-to-date dengan informasi terbaru. Dengan menggunakan data peristiwa agen, Anda dapat membuat dasbor yang menampilkan informasi dan peristiwa agen, dan memicu pemberitahuan khusus tentang aktivitas agen tertentu.
Untuk informasi selengkapnya, lihat [streaming data untuk instans Anda](https://docs.aws.amazon.com/connect/latest/adminguide/data-streaming.html), [atur ekspor waktu nyata](https://docs.aws.amazon.com/connect/latest/adminguide/set-up-real-time-export.html), dan [aliran peristiwa agen](https://docs.aws.amazon.com/connect/latest/adminguide/agent-event-streams.html) di *Panduan Administrator Amazon Connect*.
# Menulis ke Kinesis Data Streams menggunakan AWS Database Migration Service
Anda dapat menggunakan AWS Database Migration Service untuk memigrasikan data ke aliran data Kinesis. Anda dapat membangun aplikasi konsumen yang memproses catatan data secara real time. Anda juga dapat dengan mudah mengirim data hilir ke layanan lain seperti Amazon Simple Storage Service, Amazon DynamoDB, dan Amazon Redshift
Untuk informasi selengkapnya, lihat [Menggunakan Kinesis Data](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html) Streams *AWS Database Migration Service di* Panduan Pengguna.
# Menulis ke Kinesis Data Streams menggunakan Amazon DynamoDB
Gunakan Amazon Kinesis Data Streams untuk menangkap perubahan pada Amazon DynamoDB. Kinesis Data Streams menangkap modifikasi tingkat item di tabel DynamoDB apa pun dan mereplikasikannya ke Kinesis data stream. Aplikasi konsumen Anda dapat mengakses aliran ini untuk melihat perubahan tingkat item secara real time dan mengirimkan perubahan tersebut ke hilir atau mengambil tindakan berdasarkan konten.
*Untuk informasi selengkapnya, lihat [cara Kinesis Data Streams bekerja dengan](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/kds.html) DynamoDB di Panduan Pengembang Amazon DynamoDB.*
# Menulis ke Kinesis Data Streams menggunakan Amazon EventBridge
Menggunakan Kinesis Data Streams, Anda AWS dapat mengirim EventBridge peristiwa panggilan [API ke](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-events.html) aliran, membangun aplikasi konsumen, dan memproses data dalam jumlah besar. Anda juga dapat menggunakan Kinesis Data Streams sebagai EventBridge target di Pipes dan mengirimkan rekaman aliran dari salah satu sumber yang tersedia setelah pemfilteran dan pengayaan opsional.
Untuk informasi selengkapnya, lihat [Mengirim peristiwa ke aliran Amazon Kinesis](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-relay-events-kinesis-stream.html) dan [EventBridge Pipa](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html) di * EventBridge Panduan Pengguna Amazon*.
# Menulis ke Kinesis Data Streams menggunakan AWS IoT Core
Anda dapat menulis data secara real time dari pesan MQTT di IoT AWS Core dengan menggunakan tindakan Aturan IoT. AWS Anda kemudian dapat membangun aplikasi yang memproses data, menganalisis kontennya dan menghasilkan peringatan, dan mengirimkannya ke aplikasi analitik atau AWS layanan lainnya,
Untuk informasi selengkapnya, lihat [Kinesis Data](https://docs.aws.amazon.com/iot/latest/developerguide/kinesis-rule-action.html) Streams di Panduan *AWS Pengembang Inti IoT*.
# Menulis ke Kinesis Data Streams menggunakan Amazon Relational Database Service
Anda dapat menggunakan Amazon Kinesis Data Streams untuk memantau aktivitas di instans Amazon RDS Anda. Menggunakan Aliran Aktivitas Database, Amazon RDS mendorong aktivitas ke aliran data Kinesis secara real-time. Anda kemudian dapat membangun aplikasi untuk manajemen kepatuhan yang menggunakan aktivitas ini, mengaudit mereka dan menghasilkan peringatan. Anda juga dapat menggunakan Amazon Data Firehose untuk menyimpan data.
Untuk informasi selengkapnya, lihat [Aliran Aktivitas Database](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/DBActivityStreams.html) di Panduan *Pengembang Amazon RDS*.
# Menulis ke Kinesis Data Streams menggunakanAmazon Pinpoint
Anda dapat mengatur Amazon Pinpoint untuk mengirim data peristiwa ke Amazon Kinesis Data Streams. Amazon Pinpoint dapat mengirim data acara untuk kampanye, perjalanan, dan email transaksional dan pesan SMS. Anda kemudian dapat menyerap data ke dalam aplikasi analitik atau membangun aplikasi konsumen Anda sendiri yang mengambil tindakan berdasarkan konten acara.
Untuk informasi selengkapnya, lihat [Acara Streaming](https://docs.aws.amazon.com/pinpoint/latest/developerguide/event-streams.html) di Panduan *Pengembang Amazon Pinpoint*.
# Menulis ke Kinesis Data Streams menggunakan Amazon Quantum Ledger Database (Amazon QLDB)
Anda dapat membuat aliran di Amazon QLDB yang menangkap setiap revisi dokumen yang berkomitmen pada jurnal Anda dan mengirimkan data ini ke Amazon Kinesis Data Streams secara real time. Aliran QLDB adalah aliran data yang berkelanjutan dari jurnal buku besar Anda ke sumber daya aliran data Kinesis. Kemudian, Anda dapat menggunakan platform streaming Kinesis atau Perpustakaan Klien Kinesis untuk menggunakan aliran Anda, memproses catatan data, dan menganalisis konten data. Aliran QLDB menulis data Anda ke Kinesis Data Streams dalam tiga jenis rekaman:,, dan. `control` `block summary` `revision details`
Untuk informasi selengkapnya, lihat [Streaming](https://docs.aws.amazon.com/qldb/latest/developerguide/streams.html) di Panduan pengembang *Amazon QLDB*.
# Menulis ke Kinesis Data Streams menggunakan integrasi pihak ketiga
Anda dapat menulis data ke Kinesis Data Streams menggunakan salah satu opsi pihak ketiga berikut yang terintegrasi dengan Kinesis Data Streams. Pilih opsi yang ingin Anda pelajari lebih lanjut dan temukan sumber daya serta tautan ke dokumentasi yang relevan.
**Topics**
+ [Apache Flink](using-other-services-flink.md)
+ [Lancar](using-other-services-Fluentd.md)
+ [Debezium](using-other-services-Debezium.md)
+ [Oracle GoldenGate](using-other-services-Oracle-GoldenGate.md)
+ [Kafka Connect](using-other-services-kafka-connect.md)
+ [Pengalaman Adobe](using-other-services-adobe.md)
+ [Striim](using-other-services-Striim.md)
# Apache Flink
Apache Flink adalah kerangka kerja dan mesin pemrosesan terdistribusi untuk perhitungan stateful melalui aliran data tak terbatas dan terbatas. Untuk informasi selengkapnya tentang menulis ke Kinesis Data Streams dari Apache Flink, lihat Konektor Amazon [Kinesis Data](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kinesis/) Streams.
# Lancar
Fluentd adalah pengumpul data open source untuk lapisan logging terpadu. Untuk informasi lebih lanjut tentang menulis ke Kinesis Data Streams dari Fluentd. Untuk informasi lebih lanjut, lihat [Pemrosesan aliran dengan Kinesis](https://docs.fluentd.org/how-to-guides/kinesis-stream).
# Debezium
Debezium adalah platform terdistribusi open source untuk pengambilan data perubahan. Untuk informasi selengkapnya tentang menulis ke Kinesis Data Streams dari Debezium[, lihat Streaming Perubahan Data MySQL](https://debezium.io/blog/2018/08/30/streaming-mysql-data-changes-into-kinesis/) ke Amazon Kinesis.
# Oracle GoldenGate
Oracle GoldenGate adalah produk perangkat lunak yang memungkinkan Anda untuk mereplikasi, memfilter, dan mengubah data dari satu database ke database lain. Untuk informasi lebih lanjut tentang menulis ke Kinesis Data Streams GoldenGate dari Oracle[, lihat Replikasi data ke Kinesis Data](https://blogs.oracle.com/dataintegration/post/data-replication-to-aws-kinesis-data-stream-using-oracle-goldengate) Stream menggunakan Oracle. GoldenGate
# Kafka Connect
Kafka Connect adalah alat untuk streaming data yang terukur dan andal antara Apache Kafka dan sistem lainnya. Untuk informasi lebih lanjut tentang penulisan data dari Apache Kafka ke Kinesis Data Streams, [lihat](https://github.com/awslabs/kinesis-kafka-connector) konektor Kinesis kafka.
# Pengalaman Adobe
Adobe Experience Platform memungkinkan organisasi untuk memusatkan dan menstandarisasi data pelanggan dari sistem apa pun. Ini kemudian menerapkan ilmu data dan pembelajaran mesin untuk secara dramatis meningkatkan desain dan penyampaian pengalaman yang kaya dan dipersonalisasi. Untuk informasi selengkapnya tentang penulisan data dari Adobe Experience Platform ke Kinesis Data Stream. lihat cara membuat sambungan [Amazon Kinesis](https://experienceleague.adobe.com/docs/experience-platform/destinations/catalog/cloud-storage/amazon-kinesis.html?lang=en).
# Striim
Striim adalah platform lengkap dalam memori untuk mengumpulkan, memfilter, mengubah, memperkaya, menggabungkan, menganalisis, dan mengirimkan data secara real time. end-to-end [Untuk informasi lebih lanjut tentang cara menulis data ke Kinesis Data Streams dari Striim, lihat Penulis Kinesis.](https://www.striim.com/docs/en/kinesis-writer.html)
# Memecahkan masalah produsen Amazon Kinesis Data Streams
**Topics**
+ [Aplikasi produser saya menulis pada tingkat yang lebih lambat dari yang diharapkan](#producer-writing-at-slower-rate)
+ [Saya menerima kesalahan izin kunci master KMS yang tidak sah](#unauthorized-kms-producer)
+ [Memecahkan masalah umum lainnya untuk produsen](#misc-troubleshooting-producer)
## Aplikasi produser saya menulis pada tingkat yang lebih lambat dari yang diharapkan
**Topics**
+ [Batas layanan terlampaui](#service-limits-exceeded)
+ [Saya ingin mengoptimalkan produser saya](#producer-optimization)
+ [Penyalahgunaan operasi `flushSync()`](#misuse-tag)
### Batas layanan terlampaui
Untuk mengetahui apakah batas layanan terlampaui, periksa apakah produsen Anda mengeluarkan pengecualian throughput dari layanan, dan validasi operasi API apa yang sedang dibatasi. Perlu diingat bahwa ada batasan berbeda berdasarkan panggilan, lihat[Kuota dan batas](service-sizes-and-limits.md). Misalnya, selain batas tingkat shard untuk menulis dan membaca yang paling umum dikenal, ada batas tingkat aliran berikut:
+ [CreateStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_CreateStream.html)
+ [DeleteStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DeleteStream.html)
+ [ListStreams](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListStreams.html)
+ [GetShardIterator](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html)
+ [MergeShards](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_MergeShards.html)
+ [DescribeStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStream.html)
+ [DescribeStreamSummary](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStreamSummary.html)
Operasi`CreateStream`,,`DeleteStream`,`ListStreams`,`GetShardIterator`, dan `MergeShards` dibatasi hingga 5 panggilan per detik. `DescribeStream`Operasi dibatasi hingga 10 panggilan per detik. `DescribeStreamSummary`Operasi dibatasi hingga 20 panggilan per detik.
Jika panggilan ini bukan masalahnya, pastikan Anda telah memilih kunci partisi yang memungkinkan Anda mendistribusikan operasi *put* secara merata di semua pecahan, dan Anda tidak memiliki kunci partisi tertentu yang menabrak batas layanan saat sisanya tidak. Ini mengharuskan Anda mengukur throughput puncak dan memperhitungkan jumlah pecahan dalam aliran Anda. Untuk informasi selengkapnya tentang mengelola aliran, lihat[Membuat dan mengelola aliran data Kinesis](working-with-streams.md).
**Tip**
Ingatlah untuk membulatkan ke kilobyte terdekat untuk perhitungan throttling throughput saat menggunakan operasi rekam tunggal [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html), sedangkan operasi multi-rekaman [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)berputar pada jumlah kumulatif catatan di setiap panggilan. Misalnya, `PutRecords` permintaan dengan 600 catatan yang berukuran 1,1 KB tidak akan dibatasi.
### Saya ingin mengoptimalkan produser saya
Sebelum Anda mulai mengoptimalkan produser Anda, selesaikan tugas-tugas utama berikut. Pertama, identifikasi throughput puncak yang Anda inginkan dalam hal ukuran rekaman dan catatan per detik. Selanjutnya, singkirkan kapasitas aliran sebagai faktor pembatas ([Batas layanan terlampaui](#service-limits-exceeded)). Jika Anda telah mengesampingkan kapasitas streaming, gunakan tips pemecahan masalah berikut dan pedoman pengoptimalan untuk dua jenis produsen umum.
**Produser Besar**
Produsen besar biasanya berjalan dari server lokal atau instans Amazon EC2. Pelanggan yang membutuhkan throughput lebih tinggi dari produsen besar biasanya peduli dengan latensi per rekaman. Strategi untuk menangani latensi meliputi hal berikut: Jika pelanggan dapat melakukan micro-batch/buffer record, gunakan [Amazon Kinesis Producer Library](https://docs.aws.amazon.com/kinesis/latest/dev/developing-producers-with-kpl.html) (yang memiliki logika agregasi lanjutan), operasi multi-record [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html), atau agregat record ke dalam file yang lebih besar sebelum menggunakan operasi rekaman tunggal. [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html) Jika Anda tidak dapat melakukan batch atau buffer, gunakan beberapa thread untuk menulis ke layanan Kinesis Data Streams secara bersamaan. Yang AWS SDK untuk Java dan lainnya SDKs termasuk klien async yang dapat melakukan ini dengan kode yang sangat sedikit.
**Produser Kecil**
Produsen kecil biasanya adalah aplikasi seluler, perangkat IoT, atau klien web. Jika ini adalah aplikasi seluler, kami sarankan untuk menggunakan `PutRecords` operasi atau Perekam Kinesis di Ponsel. AWS SDKs Untuk informasi selengkapnya, lihat Panduan AWS Mobile SDK for Android Memulai dan Panduan AWS Mobile SDK for iOS Memulai. Aplikasi seluler harus menangani koneksi intermiten secara inheren dan memerlukan semacam batch put, seperti. `PutRecords` Jika Anda tidak dapat melakukan batch karena alasan tertentu, lihat informasi Produsen Besar di atas. Jika produsen Anda adalah browser, jumlah data yang dihasilkan biasanya sangat kecil. Namun, Anda *menempatkan operasi put* di jalur kritis aplikasi, yang tidak kami rekomendasikan.
### Penyalahgunaan operasi `flushSync()`
Menggunakan `flushSync()` secara tidak benar dapat secara signifikan memengaruhi kinerja penulisan. `flushSync()`Operasi ini dirancang untuk skenario shutdown untuk memastikan bahwa semua catatan buffer dikirim sebelum aplikasi KPL berakhir. Jika Anda menerapkan operasi ini setelah setiap operasi penulisan, ini dapat menambahkan latensi ekstra yang substansif, sekitar 500 ms per penulisan. Pastikan bahwa Anda telah menerapkan `flushSync()` hanya untuk shutdown aplikasi untuk menghindari penundaan ekstra yang tidak perlu dalam kinerja tulis.
## Saya menerima kesalahan izin kunci master KMS yang tidak sah
Kesalahan ini terjadi ketika aplikasi produser menulis ke aliran terenkripsi tanpa izin pada kunci master KMS. Untuk menetapkan izin ke aplikasi untuk mengakses kunci KMS, lihat [Menggunakan Kebijakan Utama di KMS dan [Menggunakan Kebijakan IAM](https://docs.aws.amazon.com/kms/latest/developerguide/iam-policies.html) dengan AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/key-policies.html). AWS
## Memecahkan masalah umum lainnya untuk produsen
+ [Mengapa aliran data Kinesis saya mengembalikan Kesalahan Server Internal 500?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-500-error/)
+ [Bagaimana cara mengatasi kesalahan batas waktu saat menulis dari Flink ke Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/)
+ [Bagaimana cara memecahkan masalah kesalahan pelambatan di Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling-errors/)
+ [Mengapa aliran data Kinesis saya melambat?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling/)
+ [Bagaimana saya bisa memasukkan catatan data ke dalam aliran data Kinesis menggunakan KPL?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-kpl/)
# Optimalkan produsen Kinesis Data Streams
Anda dapat mengoptimalkan lebih lanjut produsen Amazon Kinesis Data Streams tergantung pada perilaku tertentu yang Anda lihat. Tinjau topik berikut untuk mengidentifikasi solusi.
**Topics**
+ [Kustomisasi percobaan ulang KPL dan perilaku batas nilai](kinesis-producer-adv-retries-rate-limiting.md)
+ [Menerapkan praktik terbaik untuk agregasi KPL](kinesis-producer-adv-aggregation.md)
# Kustomisasi percobaan ulang KPL dan perilaku batas nilai
Saat Anda menambahkan data pengguna Amazon Kinesis Producer Library (KPL) menggunakan `addUserRecord()` operasi KPL, catatan akan diberikan stempel waktu dan ditambahkan ke buffer dengan tenggat waktu yang ditetapkan oleh parameter konfigurasi. `RecordMaxBufferedTime` stamp/deadline Kombinasi kali ini menetapkan prioritas buffer. Rekaman dikeluarkan dari buffer berdasarkan kriteria berikut:
+ Prioritas penyangga
+ Konfigurasi agregasi
+ Konfigurasi koleksi
Parameter konfigurasi agregasi dan koleksi yang memengaruhi perilaku buffer adalah sebagai berikut:
+ `AggregationMaxCount`
+ `AggregationMaxSize`
+ `CollectionMaxCount`
+ `CollectionMaxSize`
Rekaman yang di-flush kemudian dikirim ke aliran data Kinesis Anda saat Amazon Kinesis Data Streams merekam menggunakan panggilan ke operasi API Kinesis Data Streams. `PutRecords` `PutRecords`Operasi mengirimkan permintaan ke aliran Anda yang terkadang menunjukkan kegagalan penuh atau sebagian. Catatan yang gagal secara otomatis ditambahkan kembali ke buffer KPL. Batas waktu baru ditetapkan berdasarkan minimum dari dua nilai ini:
+ Setengah dari `RecordMaxBufferedTime` konfigurasi saat ini
+ time-to-liveNilai catatan
Strategi ini memungkinkan data pengguna KPL yang dicoba ulang untuk dimasukkan dalam panggilan API Kinesis Data Streams berikutnya, untuk meningkatkan throughput dan mengurangi kompleksitas sambil menerapkan nilai rekaman Kinesis Data Streams. time-to-live Tidak ada algoritma backoff, menjadikannya strategi coba lagi yang relatif agresif. Spamming karena percobaan ulang yang berlebihan dicegah dengan pembatasan tarif, dibahas di bagian selanjutnya.
## Pembatasan tarif
KPL mencakup fitur pembatas laju, yang membatasi throughput per shard yang dikirim dari satu produsen. Pembatasan laju diimplementasikan menggunakan algoritma token bucket dengan bucket terpisah untuk catatan Kinesis Data Streams dan byte. Setiap penulisan yang berhasil ke aliran data Kinesis menambahkan token (atau beberapa token) ke setiap bucket, hingga ambang batas tertentu. Ambang batas ini dapat dikonfigurasi tetapi secara default ditetapkan 50 persen lebih tinggi dari batas pecahan yang sebenarnya, untuk memungkinkan saturasi pecahan dari satu produsen.
Anda dapat menurunkan batas ini untuk mengurangi spam karena percobaan ulang yang berlebihan. Namun, praktik terbaik adalah bagi setiap produsen untuk mencoba lagi untuk throughput maksimum secara agresif dan untuk menangani setiap pelambatan yang dihasilkan ditentukan sebagai berlebihan dengan memperluas kapasitas aliran dan menerapkan strategi kunci partisi yang sesuai.
# Menerapkan praktik terbaik untuk agregasi KPL
Meskipun skema nomor urut dari catatan Amazon Kinesis Data Streams yang dihasilkan tetap sama, agregasi menyebabkan pengindeksan data pengguna Amazon Kinesis Producer Library (KPL) yang terkandung dalam rekaman Aliran Data Kinesis agregat dimulai dari 0 (nol); namun, selama Anda tidak bergantung pada nomor urut untuk mengidentifikasi catatan pengguna KPL Anda secara unik, catatan pengguna KPL Anda kode dapat mengabaikan ini, sebagai agregasi (catatan pengguna KPL Anda ke dalam catatan Kinesis Data Streams) dan de-agregasi berikutnya (dari catatan Kinesis Data Streams ke dalam catatan pengguna KPL Anda) secara otomatis mengurus ini untuk Anda. Ini berlaku apakah konsumen Anda menggunakan KCL atau AWS SDK. Untuk menggunakan fungsionalitas agregasi ini, Anda harus menarik bagian Java dari KPL ke dalam build jika konsumen Anda ditulis menggunakan API yang disediakan di SDK. AWS
Jika Anda bermaksud menggunakan nomor urut sebagai pengidentifikasi unik untuk catatan pengguna KPL Anda, kami sarankan Anda menggunakan patuh kontrak `public int hashCode()` dan `public boolean equals(Object obj)` operasi yang disediakan dalam `Record` dan `UserRecord` untuk memungkinkan perbandingan catatan pengguna KPL Anda. Selain itu, jika Anda ingin memeriksa nomor urutan catatan pengguna KPL Anda, Anda dapat mentransmisikannya ke sebuah `UserRecord` instance dan mengambil nomor urutannya.
Lihat informasi yang lebih lengkap di [Menerapkan de-agregasi konsumen](kinesis-kpl-consumer-deaggregation.md).