Informasi KCL 1.x dan 2.x

KCL 1.x

Untuk informasi selengkapnya, lihat Kembangkan konsumen KCL 1.x

KCL 2.x

Untuk informasi selengkapnya, lihat Kembangkan Konsumen KCL 2.x

Aplikasi konsumen KCL — aplikasi yang dibuat khusus menggunakan KCL dan dirancang untuk membaca dan memproses catatan dari aliran data.

Instans aplikasi konsumen - Aplikasi konsumen KCL biasanya didistribusikan, dengan satu atau lebih instance aplikasi berjalan secara bersamaan untuk mengoordinasikan kegagalan dan memproses data keseimbangan beban secara dinamis.

Wor ker — kelas tingkat tinggi yang digunakan instans aplikasi konsumen KCL untuk mulai memproses data.

Pekerja menginisialisasi dan mengawasi berbagai tugas, termasuk menyinkronkan informasi pecahan dan sewa, melacak penugasan pecahan, dan memproses data dari pecahan. Seorang pekerja menyediakan KCL dengan informasi konfigurasi untuk aplikasi konsumen, seperti nama aliran data yang rekaman datanya aplikasi konsumen KCL ini akan diproses dan AWS kredenSIAL yang diperlukan untuk mengakses aliran data ini. Pekerja juga memulai instance aplikasi konsumen KCL tertentu untuk mengirimkan catatan data dari aliran data ke prosesor catatan.

Sewa — data yang mendefinisikan ikatan antara pekerja dan pecahan. Aplikasi konsumen KCL terdistribusi menggunakan sewa untuk mempartisi pemrosesan catatan data di seluruh armada pekerja. Pada waktu tertentu, setiap pecahan catatan data terikat pada pekerja tertentu dengan sewa yang diidentifikasi oleh variabel leaseKey.

Secara default, seorang pekerja dapat memegang satu atau lebih sewa (tergantung pada nilai LeasesForWorker variabel maks) pada saat yang sama.

Misalnya, jika Anda memiliki instance aplikasi konsumen A dengan pekerja A yang memproses aliran data dengan 4 pecahan, pekerja A dapat menahan sewa ke pecahan 1, 2, 3, dan 4 secara bersamaan. Tetapi jika Anda memiliki dua instance aplikasi konsumen: A dan B dengan pekerja A dan pekerja B, dan instans ini memproses aliran data dengan 4 pecahan, pekerja A dan pekerja B tidak dapat menahan sewa ke shard 1 secara bersamaan. Seorang pekerja memegang sewa untuk pecahan tertentu sampai siap untuk berhenti memproses catatan data pecahan ini atau sampai gagal. Ketika satu pekerja berhenti memegang sewa, pekerja lain mengambil dan menahan sewa.

Untuk informasi lebih lanjut, (ini adalah repositori Java KCL), lihat https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java untuk KCL 1.x dan https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java untuk KCL 2.x.

Tabel sewa - tabel Amazon DynamoDB unik yang digunakan untuk melacak pecahan dalam aliran data KDS yang sedang disewa dan diproses oleh pekerja aplikasi konsumen KCL. Tabel sewa harus tetap sinkron (dalam pekerja dan di semua pekerja) dengan informasi shard terbaru dari aliran data saat aplikasi konsumen KCL sedang berjalan. Untuk informasi selengkapnya, lihat Gunakan tabel sewa untuk melacak pecahan yang diproses oleh aplikasi konsumen KCL.

Prosesor rekaman — logika yang mendefinisikan bagaimana aplikasi konsumen KCL Anda memproses data yang didapat dari aliran data. Saat runtime, instance aplikasi konsumen KCL membuat instance pekerja, dan pekerja ini membuat instance satu prosesor rekaman untuk setiap pecahan yang disewakan.

pos pemeriksaan: Nomor urutan pos pemeriksaan terbaru untuk pecahan. Nilai ini unik di semua pecahan dalam aliran data.

checkpointSubSequenceNumber: Saat menggunakan fitur agregasi Kinesis Producer Library, ini adalah ekstensi ke pos pemeriksaan yang melacak catatan pengguna individu dalam catatan Kinesis.

LeaseCounter: Dig unakan untuk pembuatan versi sewa sehingga pekerja dapat mendeteksi bahwa sewa mereka telah diambil oleh pekerja lain.

LeaseKey: Pengi dentifikasi unik untuk sewa. Setiap sewa khusus untuk pecahan dalam aliran data dan dipegang oleh satu pekerja pada satu waktu.

LeaseOwner: Pekerja yang memegang sewa ini.

pemilikSwitchesSinceCheckpoint: Berapa kali sewa ini telah mengubah pekerja sejak terakhir kali pos pemeriksaan ditulis.

parentShardId: Digunakan untuk memastikan bahwa pecahan induk diproses sepenuhnya sebelum pemrosesan dimulai pada pecahan anak. Ini memastikan bahwa catatan diproses dalam urutan yang sama dengan yang dimasukkan ke dalam aliran.

hashrange: Dig unakan oleh PeriodicShardSyncManager untuk menjalankan sinkronisasi berkala untuk menemukan pecahan yang hilang dalam tabel sewa dan membuat sewa untuk mereka jika diperlukan.

childshards: Digunakan oleh LeaseCleanupManager untuk meninjau status pemrosesan pecahan anak dan memutuskan apakah pecahan induk dapat dihapus dari tabel sewa.

ShardID: ID pecahan.

nama aliran Pengidentifikasi aliran data dalam format berikut:account-id:StreamName:streamCreationTimestamp.

Mengambil semua pecahan untuk data aliran yang sedang diproses

Mengambil semua sewa pecahan dari tabel sewa

Menyaring setiap pecahan terbuka yang tidak memiliki sewa di tabel sewa

Mengulangi semua pecahan terbuka yang ditemukan dan untuk setiap pecahan terbuka tanpa induk terbuka:

Selama bootstraping aplikasi, jika tabel sewa kosong, KCL menggunakan opsi pemfilteran ListShard API (parameter permintaan ShardFilter opsional) untuk mengambil dan membuat sewa hanya untuk snapshot pecahan yang terbuka pada waktu yang ditentukan oleh parameter. ShardFilter ShardFilterParameter ini memungkinkan Anda untuk menyaring respons ListShards API. Satu-satunya properti yang diperlukan dari ShardFilter parameter adalahType. KCL menggunakan properti Type filter dan nilai-nilai validnya berikut ini untuk mengidentifikasi dan mengembalikan snapshot pecahan terbuka yang mungkin memerlukan sewa baru:

ShardFilterdigunakan saat membuat sewa untuk tabel sewa kosong untuk menginisialisasi sewa untuk snapshot pecahan yang ditentukan di. RetrievalConfig#initialPositionInStreamExtended

Untuk informasi selengkapnya tentang ShardFilter, lihat https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Alih-alih semua pekerja melakukan lease/shard sinkronisasi untuk menjaga tabel sewa tetap up to date dengan pecahan terbaru dalam aliran data, satu pemimpin pekerja terpilih melakukan lease/shard sinkronisasi.

KCL 2.3 menggunakan parameter ChildShards pengembalian GetRecords dan SubscribeToShard API untuk melakukan lease/shard sinkronisasi yang terjadi pada SHARD_END pecahan tertutup, memungkinkan pekerja KCL hanya membuat sewa untuk pecahan anak dari pecahan yang selesai diprosesnya. Untuk dibagikan di seluruh aplikasi konsumen, pengoptimalan lease/shard sinkronisasi ini menggunakan ChildShards parameter GetRecords API. Untuk aplikasi konsumen throughput khusus (fan-out yang disempurnakan), pengoptimalan lease/shard sinkronisasi ini menggunakan ChildShards parameter API. SubscribeToShard Lihat informasi selengkapnya di GetRecords, SubscribeToShards, dan ChildShard.

Dengan perubahan di atas, perilaku KCL bergerak dari model semua pekerja yang belajar tentang semua pecahan yang ada ke model pekerja yang hanya belajar tentang pecahan anak-anak dari pecahan yang dimiliki setiap pekerja. Oleh karena itu, selain sinkronisasi yang terjadi selama peristiwa bootstraping dan reshard aplikasi konsumen, KCL sekarang juga melakukan shard/lease pemindaian berkala tambahan untuk mengidentifikasi lubang potensial dalam tabel sewa (dengan kata lain, untuk mempelajari semua pecahan baru) untuk memastikan rentang hash lengkap dari aliran data sedang diproses dan membuat sewa untuk mereka jika diperlukan. PeriodicShardSyncManageradalah komponen yang bertanggung jawab untuk menjalankan lease/shard pemindaian berkala.

Untuk informasi lebih lanjut tentang PeriodicShardSyncManager di KCL 2.3, lihat https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java # L201-L213.

Di KCL 2.3, opsi konfigurasi baru tersedia untuk dikonfigurasi PeriodicShardSyncManager di: LeaseManagementConfig

CloudWatch Metrik baru juga sekarang dipancarkan untuk memantau kesehatanPeriodicShardSyncManager. Untuk informasi selengkapnya, lihat PeriodicShardSyncManager.

Termasuk pengoptimalan HierarchicalShardSyncer untuk hanya membuat sewa untuk satu lapisan pecahan.

Selama bootstraping aplikasi, jika tabel sewa kosong, KCL menggunakan opsi pemfilteran ListShard API (parameter permintaan ShardFilter opsional) untuk mengambil dan membuat sewa hanya untuk snapshot pecahan yang terbuka pada waktu yang ditentukan oleh parameter. ShardFilter ShardFilterParameter ini memungkinkan Anda untuk menyaring respons ListShards API. Satu-satunya properti yang diperlukan dari ShardFilter parameter adalahType. KCL menggunakan properti Type filter dan nilai-nilai validnya berikut ini untuk mengidentifikasi dan mengembalikan snapshot pecahan terbuka yang mungkin memerlukan sewa baru:

ShardFilterdigunakan saat membuat sewa untuk tabel sewa kosong untuk menginisialisasi sewa untuk snapshot pecahan yang ditentukan di. KinesisClientLibConfiguration#initialPositionInStreamExtended

Untuk informasi selengkapnya tentang ShardFilter, lihat https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Alih-alih semua pekerja melakukan lease/shard sinkronisasi untuk menjaga tabel sewa tetap up to date dengan pecahan terbaru dalam aliran data, satu pemimpin pekerja terpilih melakukan lease/shard sinkronisasi.

KCL 1.14 menggunakan parameter ChildShards pengembalian GetRecords dan SubscribeToShard API untuk melakukan lease/shard sinkronisasi yang terjadi pada SHARD_END pecahan tertutup, memungkinkan pekerja KCL hanya membuat sewa untuk pecahan anak dari pecahan yang selesai diprosesnya. Untuk informasi selengkapnya, lihat GetRecords dan ChildShard.

Dengan perubahan di atas, perilaku KCL bergerak dari model semua pekerja yang belajar tentang semua pecahan yang ada ke model pekerja yang hanya belajar tentang pecahan anak-anak dari pecahan yang dimiliki setiap pekerja. Oleh karena itu, selain sinkronisasi yang terjadi selama peristiwa bootstraping dan reshard aplikasi konsumen, KCL sekarang juga melakukan shard/lease pemindaian berkala tambahan untuk mengidentifikasi lubang potensial dalam tabel sewa (dengan kata lain, untuk mempelajari semua pecahan baru) untuk memastikan rentang hash lengkap dari aliran data sedang diproses dan membuat sewa untuk mereka jika diperlukan. PeriodicShardSyncManageradalah komponen yang bertanggung jawab untuk menjalankan lease/shard pemindaian berkala.

Ketika KinesisClientLibConfiguration#shardSyncStrategyType disetel keShardSyncStrategyType.SHARD_END, PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold digunakan untuk menentukan ambang batas untuk jumlah pemindaian berturut-turut yang berisi lubang di tabel sewa setelah itu untuk menerapkan sinkronisasi pecahan. Ketika KinesisClientLibConfiguration#shardSyncStrategyType disetel keShardSyncStrategyType.PERIODIC, leasesRecoveryAuditorInconsistencyConfidenceThreshold diabaikan.

Untuk informasi lebih lanjut tentang PeriodicShardSyncManager di KCL 1.14, lihat https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java #. L987-L999

Di KCL 1.14, opsi konfigurasi baru tersedia untuk dikonfigurasi PeriodicShardSyncManager di: LeaseManagementConfig

CloudWatch Metrik baru juga sekarang dipancarkan untuk memantau kesehatanPeriodicShardSyncManager. Untuk informasi selengkapnya, lihat PeriodicShardSyncManager.

KCL 1.14 sekarang juga mendukung pembersihan sewa yang ditangguhkan. Sewa dihapus secara asinkron LeaseCleanupManager setelah mencapaiSHARD_END, ketika pecahan telah kedaluwarsa melewati periode retensi aliran data atau ditutup sebagai hasil dari operasi resharding.

Opsi konfigurasi baru tersedia untuk dikonfigurasiLeaseCleanupManager.

Termasuk pengoptimalan KinesisShardSyncer untuk hanya membuat sewa untuk satu lapisan pecahan.

MultistreamTracker antarmuka

Untuk membangun aplikasi konsumen yang dapat memproses beberapa aliran pada saat yang sama, Anda harus menerapkan antarmuka baru yang disebut MultistreamTracker. Antarmuka ini mencakup streamConfigList metode yang mengembalikan daftar aliran data dan konfigurasinya untuk diproses oleh aplikasi konsumen KCL. Perhatikan bahwa aliran data yang sedang diproses dapat diubah selama runtime aplikasi konsumen. streamConfigListdipanggil secara berkala oleh KCL untuk mempelajari tentang perubahan aliran data untuk diproses.

streamConfigListMetode mengisi StreamConfig daftar.

package software.amazon.kinesis.common;

import lombok.Data;
import lombok.experimental.Accessors;

@Data
@Accessors(fluent = true)
public class StreamConfig {
    private final StreamIdentifier streamIdentifier;
    private final InitialPositionInStreamExtended initialPositionInStreamExtended;
    private String consumerArn;
}
                

Perhatikan bahwa kolom StreamIdentifier dan InitialPositionInStreamExtended wajib diisi, sementara consumerArn bersifat opsional. Anda harus menyediakan consumerArn satu-satunya jika Anda menggunakan KCL 2.x untuk mengimplementasikan aplikasi konsumen fan-out yang disempurnakan.

Untuk informasi selengkapnyaStreamIdentifier, lihat https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java #L129. Untuk membuatStreamIdentifier, kami sarankan Anda membuat instance multistream dari streamArn dan yang tersedia di v2.5.0 dan streamCreationEpoch yang lebih baru. Di KCL v2.3 dan v2.4, yang tidak mendukungstreamArm, buat instance multistream dengan menggunakan format. account-id:StreamName:streamCreationTimestamp Format ini tidak akan digunakan lagi dan tidak lagi didukung mulai dengan rilis besar berikutnya.

MultistreamTrackerjuga mencakup strategi untuk menghapus sewa aliran lama di tabel sewa (formerStreamsLeasesDeletionStrategy). Perhatikan bahwa strategi TIDAK DAPAT diubah selama runtime aplikasi konsumen. Untuk informasi selengkapnya, lihat https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy.java

ConfigsBuilderadalah kelas seluruh aplikasi yang dapat Anda gunakan untuk menentukan semua pengaturan konfigurasi KCL 2.x yang akan digunakan saat membangun aplikasi konsumen KCL Anda. ConfigsBuilderkelas sekarang memiliki dukungan untuk MultistreamTracker antarmuka. Anda dapat mengin ConfigsBuilder isialisasi salah satu dengan nama satu aliran data untuk menggunakan catatan dari:

 /**
     * Constructor to initialize ConfigsBuilder with StreamName
     * @param streamName
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull String streamName, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.right(streamName);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }

Atau Anda dapat menginisialisasi ConfigsBuilder dengan MultiStreamTracker jika Anda ingin mengimplementasikan aplikasi konsumen KCL yang memproses beberapa aliran secara bersamaan.

* Constructor to initialize ConfigsBuilder with MultiStreamTracker
     * @param multiStreamTracker
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull MultiStreamTracker multiStreamTracker, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.left(multiStreamTracker);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }                    
                

Dengan dukungan multistream yang diterapkan untuk aplikasi konsumen KCL Anda, setiap baris tabel sewa aplikasi sekarang berisi ID pecahan dan nama aliran dari beberapa aliran data yang diproses aplikasi ini.

Ketika dukungan multistream untuk aplikasi konsumen KCL Anda diimplementasikan, LeaseKey mengambil struktur berikut:. account-id:StreamName:streamCreationTimestamp:ShardId Misalnya, 111111111:multiStreamTest-1:12345:shardId-000000000336.