View a markdown version of this page

Mengembangkan konsumen Kinesis Client Library di.NET - Amazon Kinesis Data Streams
Menerapkan metode IRecordProcessor kelasUbah properti konfigurasi

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Mengembangkan konsumen Kinesis Client Library di.NET

penting

Amazon Kinesis Client Library (KCL) versi 1.x dan 2.x sudah usang. KCL 1.x akan mencapai akhir dukungan pada 30 Januari 2026. Kami sangat menyarankan Anda memigrasikan aplikasi KCL menggunakan versi 1.x ke versi KCL terbaru sebelum 30 Januari 2026. Untuk menemukan versi KCL terbaru, lihat halaman Perpustakaan Klien Amazon Kinesis di. GitHub Untuk informasi tentang versi KCL terbaru, lihatGunakan Perpustakaan Klien Kinesis. Untuk informasi tentang migrasi dari KCL 1.x ke KCL 3.x, lihat. Migrasi dari KCL 1.x ke KCL 3.x

Anda dapat menggunakan Kinesis Client Library (KCL) untuk membangun aplikasi yang memproses data dari aliran data Kinesis Anda. Perpustakaan Klien Kinesis tersedia dalam berbagai bahasa. Topik ini membahas .NET.

KCL adalah perpustakaan Java; dukungan untuk bahasa selain Java disediakan menggunakan antarmuka multi-bahasa yang disebut. MultiLangDaemon Daemon ini Java-based dan berjalan di latar belakang ketika Anda menggunakan bahasa KCL selain Java. Oleh karena itu, jika Anda menginstal KCL untuk .NET dan menulis aplikasi konsumen Anda sepenuhnya di.NET, Anda masih perlu menginstal Java di sistem Anda karena. MultiLangDaemon Selanjutnya, MultiLangDaemon memiliki beberapa pengaturan default yang mungkin perlu Anda sesuaikan untuk kasus penggunaan Anda, misalnya, AWS Wilayah yang terhubung dengannya. Untuk informasi lebih lanjut MultiLangDaemon tentang aktif GitHub, buka halaman MultiLangDaemon proyek KCL.

Untuk mengunduh.NET KCL dari GitHub, buka Kinesis Client Library (.NET). Untuk mengunduh kode sampel untuk aplikasi konsumen.NET KCL, buka halaman proyek konsumen sampel KCL untuk .NET di. GitHub

Anda harus menyelesaikan tugas-tugas berikut saat mengimplementasikan aplikasi konsumen KCL di.NET:

Menerapkan metode IRecordProcessor kelas

Konsumen harus menerapkan metode berikut untukIRecordProcessor. Konsumen sampel menyediakan implementasi yang dapat Anda gunakan sebagai titik awal (lihat SampleRecordProcessor kelas diSampleConsumer/AmazonKinesisSampleConsumer.cs).

public void Initialize(InitializationInput input)
public void ProcessRecords(ProcessRecordsInput input)
public void Shutdown(ShutdownInput input)
Inisialisasi

KCL memanggil metode ini ketika prosesor rekaman dibuat instansiasi, meneruskan ID pecahan tertentu dalam parameter (). input input.ShardId Prosesor rekaman ini hanya memproses pecahan ini, dan biasanya, kebalikannya juga benar (pecahan ini hanya diproses oleh prosesor rekaman ini). Namun, konsumen Anda harus memperhitungkan kemungkinan bahwa catatan data mungkin diproses lebih dari satu kali. Ini karena Kinesis Data Streams memiliki setidaknya sekali semantik, yang berarti bahwa setiap catatan data dari pecahan diproses setidaknya satu kali oleh pekerja di konsumen Anda. Untuk informasi selengkapnya tentang kasus di mana pecahan tertentu mungkin diproses oleh lebih dari satu pekerja, lihatGunakan resharding, scaling, dan parallel processing untuk mengubah jumlah pecahan.

public void Initialize(InitializationInput input)
ProcessRecords

KCL memanggil metode ini, meneruskan daftar catatan data dalam input parameter (input.Records) dari pecahan yang ditentukan oleh metode. Initialize Prosesor catatan yang Anda terapkan memproses data dalam catatan ini sesuai dengan semantik konsumen Anda. Misalnya, pekerja mungkin melakukan transformasi pada data dan kemudian menyimpan hasilnya dalam bucket Amazon Simple Storage Service (Amazon S3).

public void ProcessRecords(ProcessRecordsInput input)

Selain data itu sendiri, catatan juga berisi nomor urut dan kunci partisi. Pekerja dapat menggunakan nilai-nilai ini saat memproses data. Misalnya, pekerja dapat memilih bucket S3 untuk menyimpan data berdasarkan nilai kunci partisi. Kel Record as mengekspos berikut ini untuk mengakses data rekaman, nomor urut, dan kunci partisi:

byte[] Record.Data 
string Record.SequenceNumber
string Record.PartitionKey

Dalam sampel, metode ProcessRecordsWithRetries memiliki kode yang menunjukkan bagaimana pekerja dapat mengakses data catatan, nomor urut, dan kunci partisi.

Kinesis Data Streams membutuhkan prosesor rekaman untuk melacak catatan yang telah diproses dalam pecahan. KCL menangani pelacakan ini untuk Anda dengan meneruskan Checkpointer objek ke ProcessRecords (input.Checkpointer). Prosesor rekaman memang Checkpointer.Checkpoint gil metode untuk menginformasikan KCL tentang seberapa jauh kemajuannya dalam memproses catatan dalam pecahan. Jika pekerja gagal, KCL menggunakan informasi ini untuk memulai ulang pemrosesan pecahan pada catatan terakhir yang diproses yang diketahui.

Untuk operasi split atau merger, KCL tidak mulai memproses pecahan baru sampai prosesor untuk pecahan asli telah memanggil untuk memberi sinyal bahwa semua pemrosesan pada pecahan asli selesai. Checkpointer.Checkpoint

Jika Anda tidak meneruskan parameter, KCL mengasumsikan bahwa panggilan untuk Checkpointer.Checkpoint menandakan bahwa semua catatan telah diproses, hingga catatan terakhir yang diteruskan ke prosesor rekaman. Oleh karena itu, prosesor rekaman harus memanggil Checkpointer.Checkpoint hanya setelah memproses semua catatan dalam daftar yang diteruskan kepadanya. Prosesor rekaman tidak perlu menelep Checkpointer.Checkpoint on setiap panggilan keProcessRecords. Prosesor dapat, misalnya, menelep Checkpointer.Checkpoint on setiap panggilan ketiga atau keempat. Anda dapat secara opsional menentukan nomor urutan yang tepat dari catatan sebagai parameter untukCheckpointer.Checkpoint. Dalam hal ini, KCL mengasumsikan bahwa catatan telah diproses hanya hingga catatan itu.

Dalam sampel, metode pribadi Checkpoint(Checkpointer checkpointer) menunjukkan cara memanggil Checkpointer.Checkpoint metode menggunakan penanganan pengecualian yang sesuai dan logika coba ulang.

KCL untuk .NET menangani pengecualian secara berbeda dari pustaka bahasa KCL lainnya karena tidak menangani pengecualian apa pun yang muncul dari pemrosesan catatan data. Setiap pengecualian yang tidak tertangkap dari kode pengguna membuat program rusak.

Matikan

KCL memanggil Shutdown metode baik ketika pemrosesan berakhir (alasan shutdown adalahTERMINATE) atau pekerja tidak lagi merespons (input.Reasonnilai shutdown adalahZOMBIE).

public void Shutdown(ShutdownInput input)

Pemrosesan berakhir ketika prosesor rekaman tidak menerima catatan lebih lanjut dari pecahan, karena pecahan dipecah atau digabungkan, atau aliran dihapus.

KCL juga meneruskan Checkpointer objek keshutdown. Jika alasan shutdown adalahTERMINATE, prosesor rekaman harus menyelesaikan pemrosesan catatan data apa pun, dan kemudian memanggil checkpoint metode pada antarmuka ini.

Ubah properti konfigurasi

Konsumen sampel memberikan nilai default untuk properti konfigurasi. Anda dapat mengganti salah satu properti ini dengan nilai Anda sendiri (lihatSampleConsumer/kcl.properties).

Nama aplikasi

KCL memerlukan aplikasi yang unik di antara aplikasi Anda, dan di antara tabel Amazon DynamoDB di Wilayah yang sama. Ini menggunakan nilai konfigurasi nama aplikasi dengan cara berikut:

  • Semua pekerja yang terkait dengan nama aplikasi ini diasumsikan bekerja bersama pada aliran yang sama. Pekerja ini dapat didistribusikan pada beberapa contoh. Jika Anda menjalankan instance tambahan dari kode aplikasi yang sama, tetapi dengan nama aplikasi yang berbeda, KCL memperlakukan instance kedua sebagai aplikasi yang sepenuhnya terpisah yang juga beroperasi pada aliran yang sama.

  • KCL membuat tabel DynamoDB dengan nama aplikasi dan menggunakan tabel untuk mempertahankan informasi status (seperti pos pemeriksaan dan pemetaan pecahan pekerja) untuk aplikasi. Setiap aplikasi memiliki tabel DynamoDB sendiri. Untuk informasi selengkapnya, lihat Gunakan tabel sewa untuk melacak pecahan yang diproses oleh aplikasi konsumen KCL.

Menyiapkan kredensi

Anda harus membuat AWS kredenSIAL Anda tersedia untuk salah satu penyedia kredensia dalam rantai penyedia kredensia default. Anda dapat menggunakan AWSCredentialsProvider properti untuk menetapkan penyedia kredensibilitas. Sample.properties harus membuat kredenSIAL Anda tersedia untuk salah satu penyedia kredenSIAL dalam rantai penyedia kreden sia default. Jika Anda menjalankan aplikasi konsumen pada instans EC2, sebaiknya Anda mengonfigurasi instans dengan peran IAM. AWS kredenSIAL yang mencerminkan izin yang terkait dengan peran IAM ini tersedia untuk aplikasi pada instans melalui metadata instansnya. Ini adalah cara paling aman untuk mengelola kredenSIAL bagi konsumen yang menjalankan instans EC2.

File properti sampel mengonfigurasi KCL untuk memproses aliran data Kinesis yang disebut “kata-kata” menggunakan prosesor rekaman yang disediakan. AmazonKinesisSampleConsumer.cs