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

Kembangkan konsumen Kinesis Client Library di Node.js

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 Node.js.

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 Node.js dan menulis aplikasi konsumen Anda sepenuhnya Node.js, 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 Node.js KCL dari GitHub, buka Kinesis Client Library ()Node.js.

Anda harus menyelesaikan tugas-tugas berikut saat menerapkan aplikasi konsumen KCL di: Node.js

Menerapkan prosesor rekaman

Konsumen paling sederhana yang menggunakan KCL untuk Node.js harus mengimplementasikan recordProcessor fungsi, yang pada gilirannya berisi fungsiinitialize,processRecords, danshutdown. Sampel menyediakan implementasi yang dapat Anda gunakan sebagai titik awal (lihatsample_kcl_app.js).

function recordProcessor() {
  // return an object that implements initialize, processRecords and shutdown functions.}

menginisialisasi

KCL memanggil initialize fungsi saat prosesor rekaman dimulai. Prosesor rekaman ini hanya memproses ID pecahan yang diter initializeInput.shardId uskan sebagai, 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.

initialize: function(initializeInput, completeCallback)

ProsesRecord

KCL memanggil fungsi ini dengan input yang berisi daftar catatan data dari pecahan yang ditentukan ke initialize fungsi. 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).

processRecords: function(processRecordsInput, completeCallback)

Selain data itu sendiri, catatan juga berisi nomor urut dan kunci partisi, yang dapat digunakan pekerja saat memproses data. Misalnya, pekerja dapat memilih bucket S3 untuk menyimpan data berdasarkan nilai kunci partisi. Kam record us menampilkan pasangan kunci-nilai berikut untuk mengakses data catatan, nomor urut, dan kunci partisi:

record.data
record.sequenceNumber
record.partitionKey

Perhatikan bahwa datanya adalah Base64-encoded.

Dalam contoh dasar, fungsi processRecords 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 dengan checkpointer objek yang diteruskan sebagaiprocessRecordsInput.checkpointer. Prosesor rekaman Anda checkpointer.checkpoint memanggil fungsi untuk memberi tahu KCL seberapa jauh perkembangannya dalam memproses catatan di pecahan. Jika pekerja gagal, KCL menggunakan informasi ini ketika Anda memulai ulang pemrosesan pecahan sehingga berlanjut dari catatan pemrosesan terakhir 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. checkpoint

Jika Anda tidak meneruskan nomor urut ke checkpoint fungsi, KCL mengasumsikan bahwa panggilan ke checkpoint berarti bahwa semua catatan telah diproses, hingga catatan terakhir yang diteruskan ke prosesor rekaman. Oleh karena itu, prosesor rekaman harus memanggil checkpoint hanya setelah memproses semua catatan dalam daftar yang diteruskan kepadanya. Prosesor rekaman tidak perlu menelep checkpoint on setiap panggilan keprocessRecords. Prosesor dapat, misalnya, memang checkpoint gil setiap panggilan ketiga, atau beberapa peristiwa di luar prosesor rekaman Anda, seperti verification/validation layanan khusus yang telah Anda terapkan.

Anda dapat secara opsional menentukan nomor urutan yang tepat dari catatan sebagai parameter untukcheckpoint. Dalam hal ini, KCL mengasumsikan bahwa semua catatan telah diproses hingga catatan itu saja.

Aplikasi sampel dasar menunjukkan panggilan paling sederhana yang mungkin untuk checkpointer.checkpoint fungsi tersebut. Anda dapat menambahkan logika checkpointing lain yang Anda butuhkan untuk konsumen Anda pada titik ini dalam fungsi.

penonaktifan

KCL memanggil shutdown fungsi baik ketika pemrosesan berakhir (shutdownInput.reasonisTERMINATE) atau pekerja tidak lagi merespons (shutdownInput.reasonisZOMBIE).

shutdown: function(shutdownInput, completeCallback)

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

KCL juga meneruskan shutdownInput.checkpointer objek keshutdown. Jika alasan shutdown adalahTERMINATE, Anda harus memastikan bahwa prosesor rekaman telah selesai memproses catatan data apa pun, dan kemudian memanggil checkpoint fungsi pada antarmuka ini.

Ubah properti konfigurasi

Sampel memberikan nilai default untuk properti konfigurasi. Anda dapat mengganti salah satu properti ini dengan nilai Anda sendiri (lihat sample.properties dalam sampel dasar).

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:

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.propertiesFile harus membuat kredenSIAL Anda tersedia untuk salah satu penyedia kredenSIAL dalam rantai penyedia kredensia default. Jika Anda menjalankan konsumen di instans Amazon 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 untuk aplikasi konsumen yang berjalan pada instans EC2.

Contoh berikut mengonfigurasi KCL untuk memproses aliran data Kinesis bernama kclnodejssample menggunakan prosesor rekaman yang disediakan di: sample_kcl_app.js

# The Node.js executable script
executableName = node sample_kcl_app.js
# The name of an Amazon Kinesis stream to process
streamName = kclnodejssample
# Unique KCL application name
applicationName = kclnodejssample
# Use default AWS credentials provider chain
AWSCredentialsProvider = DefaultAWSCredentialsProviderChain
# Read from the beginning of the stream
initialPositionInStream = TRIM_HORIZON