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 Java

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 tentang Java. Untuk melihat referensi Javadoc, lihat topik AWS Javadoc untuk Kelas. AmazonKinesisClient

Untuk mengunduh Java KCL dari GitHub, buka Kinesis Client Library (Java). Untuk menemukan Java KCL di Apache Maven, buka halaman hasil pencarian KCL. Untuk mengunduh kode sampel untuk aplikasi konsumen Java KCL dari GitHub, buka halaman proyek sampel KCL untuk Java di. GitHub

Aplikasi sampel menggunakan Apache Commons Logging. Anda dapat mengubah konfigurasi logging dalam configure metode statis yang ditentukan dalam AmazonKinesisApplicationSample.java file. Untuk informasi selengkapnya tentang cara menggunakan Apache Commons Logging dengan aplikasi Log4j dan AWS Java, lihat Log ging dengan Log4j di Panduan Pengembang. AWS SDK untuk Java

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

Menerapkan IRecordProcessor metode

KCL saat ini mendukung dua versi antarmuka: IRecordProcessor Antarmuka asli tersedia dengan versi pertama KCL, dan versi 2 tersedia mulai dengan KCL versi 1.5.0. Kedua antarmuka didukung sepenuhnya. Pilihan Anda tergantung pada persyaratan skenario spesifik Anda. Lihat Javadocs yang dibuat secara lokal atau kode sumber untuk melihat semua perbedaannya. Bagian berikut menguraikan implementasi minimal untuk memulai.

Antarmuka Asli (Versi 1)

Antar IRecordProcessor muka asli (package com.amazonaws.services.kinesis.clientlibrary.interfaces) memperlihatkan metode pengolah rekaman berikut yang harus diterapkan oleh konsumen Anda. Sampel menyediakan implementasi yang dapat Anda gunakan sebagai titik awal (lihatAmazonKinesisApplicationSampleRecordProcessor.java).

public void initialize(String shardId)
public void processRecords(List<Record> records, IRecordProcessorCheckpointer checkpointer)
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) 

menginisialisasi

KCL memanggil initialize metode ketika prosesor rekaman dibuat instansiasi, meneruskan ID pecahan tertentu sebagai parameter. 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. 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 lebih lanjut tentang kasus di mana pecahan tertentu dapat diproses oleh lebih dari satu pekerja, lihatGunakan resharding, scaling, dan parallel processing untuk mengubah jumlah pecahan.

public void initialize(String shardId)

ProsesRecord

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

public void processRecords(List<Record> records, IRecordProcessorCheckpointer checkpointer) 

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 metode berikut yang menyediakan akses ke data catatan, nomor urut, dan kunci partisi.

record.getData()  
record.getSequenceNumber() 
record.getPartitionKey() 

Dalam sampel, metode pribadi 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 (IRecordProcessorCheckpointer) ke. processRecords Prosesor rekaman memang checkpoint gil metode pada antarmuka ini untuk menginformasikan KCL tentang seberapa jauh perkembangannya 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 akan 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 parameter, 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 keprocessRecords. 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.

Dalam sampel, metode pribadi checkpoint menunjukkan cara memanggil IRecordProcessorCheckpointer.checkpoint menggunakan penanganan pengecualian yang sesuai dan logika coba ulang.

KCL mengandalkan processRecords untuk menangani pengecualian apa pun yang timbul dari pemrosesan catatan data. Jika pengecualian dilemparkan dariprocessRecords, KCL melewatkan catatan data yang diteruskan sebelum pengecualian. Artinya, catatan ini tidak dikirim kembali ke prosesor rekaman yang melemparkan pengecualian atau ke prosesor rekaman lain di konsumen.

penonaktifan

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

public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason)

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

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

Antarmuka yang diperbarui (Versi 2)

Antar IRecordProcessor muka yang diperbarui (package com.amazonaws.services.kinesis.clientlibrary.interfaces.v2) memperlihatkan metode pengolah rekaman berikut yang harus diterapkan konsumen Anda:

void initialize(InitializationInput initializationInput)
void processRecords(ProcessRecordsInput processRecordsInput)
void shutdown(ShutdownInput shutdownInput)

Semua argumen dari versi asli antarmuka dapat diakses melalui metode get pada objek kontainer. Misalnya, untuk mengambil daftar catatan diprocessRecords(), Anda dapat menggunakanprocessRecordsInput.getRecords().

Pada versi 2 antarmuka ini (KCL 1.5.0 dan yang lebih baru), input baru berikut tersedia selain input yang disediakan oleh antarmuka asli:

Menerapkan pabrik kelas untuk IRecordProcessor antarmuka

Anda juga perlu mengimplementasikan pabrik untuk kelas yang mengimplementasikan metode rekaman prosesor. Ketika konsumen Anda membuat instance pekerja, ia meneruskan referensi ke pabrik ini.

Sampel mengimplementasikan kelas pabrik dalam file AmazonKinesisApplicationSampleRecordProcessorFactory.java menggunakan antarmuka prosesor rekaman asli. Jika Anda ingin pabrik kelas membuat prosesor rekaman versi 2, gunakan nama paketcom.amazonaws.services.kinesis.clientlibrary.interfaces.v2.

  public class SampleRecordProcessorFactory implements IRecordProcessorFactory { 
      /**
      * Constructor.
      */
      public SampleRecordProcessorFactory() {
          super();
      }
      /**
      * {@inheritDoc}
      */
      @Override
      public IRecordProcessor createProcessor() {
          return new SampleRecordProcessor();
      }
  } 

Buat pekerja

Seperti yang dibahas diMenerapkan IRecordProcessor metode, ada dua versi antarmuka prosesor rekaman KCL untuk dipilih, yang memengaruhi cara Anda membuat pekerja. Antarmuka prosesor rekaman asli menggunakan struktur kode berikut untuk membuat pekerja:

final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker(recordProcessorFactory, config);

Dengan versi 2 dari antarmuka prosesor rekaman, Anda dapat menggunakan Worker.Builder untuk membuat pekerja tanpa perlu khawatir tentang konstruktor mana yang akan digunakan dan urutan argumen. Antarmuka prosesor rekaman yang diperbarui menggunakan struktur kode berikut untuk membuat pekerja:

final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker.Builder()
    .recordProcessorFactory(recordProcessorFactory)
    .config(config)
    .build();

Ubah properti konfigurasi

Sampel memberikan nilai default untuk properti konfigurasi. Data konfigurasi untuk pekerja ini kemudian dikonsolidasikan dalam sebuah KinesisClientLibConfiguration objek. Objek ini dan referensi ke kelas factory for IRecordProcessor diteruskan dalam panggilan yang membuat instance pekerja. Anda dapat mengganti salah satu properti ini dengan nilai Anda sendiri menggunakan file properti Java (lihatAmazonKinesisApplicationSample.java).

Nama aplikasi

KCL memerlukan nama aplikasi yang unik di seluruh aplikasi Anda, dan di seluruh 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. Misalnya, jika Anda menjalankan konsumen pada instans EC2, sebaiknya luncurkan 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.

Contoh aplikasi pertama kali mencoba mengambil kredenSIAL IAM dari metadata instance:

credentialsProvider = new InstanceProfileCredentialsProvider(); 

Jika contoh aplikasi tidak dapat memperoleh kredenSIAL dari metadata instance, aplikasi akan mencoba mengambil kredenSIAL dari file properti:

credentialsProvider = new ClasspathPropertiesFileCredentialsProvider();

Untuk informasi selengkapnya tentang metadata instans, lihat Metadata Inst ans di Panduan Pengguna Amazon EC2.

Menggunakan ID pekerja untuk beberapa instans

Contoh kode inisialisasi membuat ID untuk pekerja,workerId, menggunakan nama komputer lokal dan menambahkan pengidentifikasi unik global seperti yang ditunjukkan dalam cuplikan kode berikut. Pendekatan ini mendukung skenario beberapa contoh aplikasi konsumen yang berjalan pada satu komputer.

String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID();

Bermigrasi ke Versi 2 dari antarmuka prosesor rekaman

Jika Anda ingin memigrasikan kode yang menggunakan antarmuka asli, selain langkah-langkah yang dijelaskan sebelumnya, langkah-langkah berikut diperlukan: