KCL1.x dan 2.x informasi

KCL1.x

Untuk informasi selengkapnya, silakan lihat Mengembangkan KCL 1.x konsumen

KCL2.x

Untuk informasi selengkapnya, silakan lihat Kembangkan KCL 2.x Konsumen

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

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

Worker — kelas tingkat tinggi yang digunakan instance aplikasi KCL konsumen untuk mulai memproses data.

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

Sewa — data yang mendefinisikan pengikatan antara pekerja dan pecahan. Aplikasi KCL konsumen 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 (tunduk pada nilai variabel maxLeasesForWorker) 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 menyimpan 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 instance ini memproses aliran data dengan 4 pecahan, pekerja A dan pekerja B tidak dapat menahan sewa untuk shard 1 pada saat yang bersamaan. Seorang pekerja memegang sewa ke pecahan tertentu sampai siap untuk berhenti memproses catatan data pecahan ini atau sampai gagal. Ketika satu pekerja berhenti memegang sewa, pekerja lain mengambil dan memegang sewa.

Untuk informasi lebih lanjut, (ini adalah KCL repositori Java), 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 2.x. KCL

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

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

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

checkpointSubSequenceNomor: Saat menggunakan fitur agregasi Perpustakaan Produser Kinesis, ini adalah ekstensi ke pos pemeriksaan yang melacak catatan pengguna individu dalam catatan Kinesis.

leaseCounter: Digunakan untuk pembuatan versi sewa sehingga pekerja dapat mendeteksi bahwa sewa mereka telah diambil oleh pekerja lain.

leaseKey: Pengidentifikasi 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.

ownerSwitchesSincePos pemeriksaan: Berapa kali sewa ini telah berganti 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: Digunakan oleh PeriodicShardSyncManager untuk menjalankan sinkronisasi berkala untuk menemukan pecahan yang hilang di 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 bootstrap aplikasi, jika tabel sewa kosong, KCL gunakan opsi pemfilteran (parameter permintaan ShardFilter opsional) untuk mengambil dan membuat sewa hanya untuk snapshot pecahan yang terbuka pada waktu yang ditentukan oleh parameter. ListShard API ShardFilter ShardFilterParameter memungkinkan Anda untuk menyaring respons dari ListShardsAPI. Satu-satunya properti yang diperlukan dari ShardFilter parameter adalahType. KCLmenggunakan properti Type filter dan nilai valid berikut 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 synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard sinkronisasi.

KCL2.3 menggunakan parameter ChildShards pengembalian GetRecords dan SubscribeToShard APIs untuk melakukan sinkronisasi sewa/pecahan yang terjadi pada pecahan tertutup, memungkinkan KCL pekerja SHARD_END untuk hanya membuat sewa untuk pecahan anak dari pecahan yang selesai diproses. Untuk dibagikan di seluruh aplikasi konsumen, optimalisasi sinkronisasi sewa/pecahan ini menggunakan parameter. ChildShards GetRecords API Untuk aplikasi konsumen throughput khusus (fan-out yang ditingkatkan), optimalisasi sinkronisasi sewa/pecahan ini menggunakan parameter. ChildShards SubscribeToShard API Untuk informasi lebih lanjut, lihat GetRecords, SubscribeToShards, dan ChildShard.

Dengan perubahan di atas, perilaku bergerak dari model semua pekerja yang belajar tentang semua pecahan yang ada ke model pekerja yang hanya belajar tentang pecahan pecahan anak-anak yang dimiliki setiap pekerja. KCL Oleh karena itu, selain sinkronisasi yang terjadi selama bootstraping aplikasi konsumen dan peristiwa reshard, KCL sekarang juga melakukan pemindaian shard/lease 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 pemindaian lease/shard berkala.

Untuk informasi selengkapnya 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 diLeaseManagementConfig:

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

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

Selama bootstrap aplikasi, jika tabel sewa kosong, KCL gunakan opsi pemfilteran (parameter permintaan ShardFilter opsional) untuk mengambil dan membuat sewa hanya untuk snapshot pecahan yang terbuka pada waktu yang ditentukan oleh parameter. ListShard API ShardFilter ShardFilterParameter memungkinkan Anda untuk menyaring respons dari ListShardsAPI. Satu-satunya properti yang diperlukan dari ShardFilter parameter adalahType. KCLmenggunakan properti Type filter dan nilai valid berikut 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 synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard sinkronisasi.

KCL1.14 menggunakan parameter ChildShards pengembalian GetRecords dan SubscribeToShard APIs untuk melakukan sinkronisasi sewa/pecahan yang terjadi pada pecahan tertutup, memungkinkan KCL pekerja SHARD_END untuk hanya membuat sewa untuk pecahan anak dari pecahan yang selesai diproses. Untuk informasi lebih lanjut, lihat GetRecordsdan ChildShard.

Dengan perubahan di atas, perilaku bergerak dari model semua pekerja yang belajar tentang semua pecahan yang ada ke model pekerja yang hanya belajar tentang pecahan pecahan anak-anak yang dimiliki setiap pekerja. KCL Oleh karena itu, selain sinkronisasi yang terjadi selama bootstraping aplikasi konsumen dan peristiwa reshard, KCL sekarang juga melakukan pemindaian shard/lease 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 pemindaian lease/shard berkala.

Kapan KinesisClientLibConfiguration#shardSyncStrategyType diatur keShardSyncStrategyType.SHARD_END, PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold digunakan untuk menentukan ambang batas untuk jumlah pemindaian berturut-turut yang berisi lubang di tabel sewa setelah itu untuk menegakkan sinkronisasi pecahan. Kapan KinesisClientLibConfiguration#shardSyncStrategyType diatur 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 diLeaseManagementConfig:

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

KCL1.14 sekarang juga mendukung pembersihan sewa yang ditangguhkan. Sewa dihapus secara asinkron LeaseCleanupManager pada saat 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 optimasi 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 KCL konsumen. Perhatikan bahwa aliran data yang sedang diproses dapat diubah selama runtime aplikasi konsumen. streamConfigListdisebut secara berkala oleh KCL untuk mempelajari tentang perubahan aliran data untuk diproses.

streamConfigListMetode ini mengisi StreamConfigdaftar.

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 bidang StreamIdentifier dan InitialPositionInStreamExtended wajib, sementara consumerArn adalah 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 akan usang dan tidak lagi didukung dimulai dengan rilis utama berikutnya.

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

ConfigsBuilderadalah kelas seluruh aplikasi yang dapat Anda gunakan untuk menentukan semua pengaturan konfigurasi KCL 2.x yang akan digunakan saat membangun aplikasi konsumen AndaKCL. ConfigsBuilderkelas sekarang memiliki dukungan untuk MultistreamTracker antarmuka. Anda dapat menginisialisasi ConfigsBuilder baik 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 menerapkan aplikasi KCL konsumen 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 KCL konsumen 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 KCL konsumen Anda diimplementasikan, leaseKey mengambil struktur berikut:account-id:StreamName:streamCreationTimestamp:ShardId. Misalnya, 111111111:multiStreamTest-1:12345:shardId-000000000336.