Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
# Entwickeln Sie KCL 1.x-Verbraucher
**Wichtig**
Die Versionen 1.x und 2.x der Amazon Kinesis Client Library (KCL) sind veraltet. KCL 1.x wird am 30. Januar 2026 end-of-support verfügbar sein. Wir **empfehlen dringend**, dass Sie Ihre KCL-Anwendungen, die Version 1.x verwenden, vor dem 30. Januar 2026 auf die neueste KCL-Version migrieren. Die neueste KCL-Version finden Sie auf der [Seite Amazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client) unter. GitHub Informationen zu den neuesten KCL-Versionen finden Sie unter. [Verwenden Sie die Kinesis-Clientbibliothek](kcl.md) Informationen zur Migration von KCL 1.x zu KCL 3.x finden Sie unter. [Migrieren von KCL 1.x zu KCL 3.x](kcl-migration-1-3.md)
Sie können eine Verbraucheranwendung für Amazon Kinesis Data Streams entwickeln, indem Sie die Kinesis Client Library (KCL) verwenden.
Weitere Informationen zu KCL finden Sie unter [Über KCL (frühere Versionen)](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-overview).
Wählen Sie je nach der Option, die Sie verwenden möchten, aus den folgenden Themen.
**Topics**
+ [Entwickeln Sie einen Kinesis Client Library-Consumer in Java](kinesis-record-processor-implementation-app-java.md)
+ [Entwickeln Sie einen Kinesis Client Library-Consumer in Node.js](kinesis-record-processor-implementation-app-nodejs.md)
+ [Entwickeln Sie einen Kinesis Client Library-Consumer in .NET](kinesis-record-processor-implementation-app-dotnet.md)
+ [Entwickeln Sie einen Kinesis Client Library-Consumer in Python](kinesis-record-processor-implementation-app-py.md)
+ [Entwickeln Sie einen Kinesis Client Library-Consumer in Ruby](kinesis-record-processor-implementation-app-ruby.md)
# Entwickeln Sie einen Kinesis Client Library-Consumer in Java
**Wichtig**
Die Versionen 1.x und 2.x der Amazon Kinesis Client Library (KCL) sind veraltet. KCL 1.x wird end-of-support am 30. Januar 2026 verfügbar sein. Wir **empfehlen dringend**, dass Sie Ihre KCL-Anwendungen, die Version 1.x verwenden, vor dem 30. Januar 2026 auf die neueste KCL-Version migrieren. Die neueste KCL-Version finden Sie auf der [Seite Amazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client) unter. GitHub Informationen zu den neuesten KCL-Versionen finden Sie unter. [Verwenden Sie die Kinesis-Clientbibliothek](kcl.md) Informationen zur Migration von KCL 1.x zu KCL 3.x finden Sie unter. [Migrieren von KCL 1.x zu KCL 3.x](kcl-migration-1-3.md)
Sie können die Kinesis Client Library (KCL) verwenden, um Anwendungen zu erstellen, die Daten aus Ihren Kinesis-Datenströmen verarbeiten. Die Kinesis Client Library ist in mehreren Sprachen verfügbar. In diesem Thema wird Java behandelt. [Die Javadoc-Referenz finden Sie im Javadoc-Thema für Class.AWS AmazonKinesisClient](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/AmazonKinesisClient.html)
Gehen Sie zur [Kinesis Client Library (Java) GitHub, um die Java-KCL](https://github.com/awslabs/amazon-kinesis-client) von herunterzuladen. Um die Java KCL auf Apache Maven zu finden, navigieren Sie zur Seite für die [KCL-Suchergebnisse](https://search.maven.org/#search|ga|1|amazon-kinesis-client). Um Beispielcode für eine Java-KCL-Verbraucheranwendung herunterzuladen GitHub, gehen Sie auf die Projektseite [KCL for Java-Beispielprojekt unter](https://github.com/aws/aws-sdk-java/tree/master/src/samples/AmazonKinesis). GitHub
Die Beispielanwendung verwendet [Apache Commons Logging](http://commons.apache.org/proper/commons-logging/guide.html). Sie können die Konfiguration der Protokollierung in der statischen Methode `configure` ändern, die in der Datei `AmazonKinesisApplicationSample.java` definiert ist. *Weitere Informationen zur Verwendung von Apache Commons Logging mit Log4j und AWS Java-Anwendungen finden Sie unter [Logging with Log4j](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/java-dg-logging.html) im Developer Guide.AWS SDK für Java *
Sie müssen die folgenden Aufgaben durchführen, wenn Sie eine KCL-Konsumentenanwendung in Java implementieren:
**Topics**
+ [Implementieren Sie die Methoden des Prozessors IRecord](#kinesis-record-processor-implementation-interface-java)
+ [Implementieren Sie eine Klassenfabrik für die IRecord Prozessorschnittstelle](#kinesis-record-processor-implementation-factory-java)
+ [Erstellen Sie einen Worker](#kcl-java-worker)
+ [Ändern Sie die Konfigurationseigenschaften](#kinesis-record-processor-initialization-java)
+ [Migrieren Sie zu Version 2 der Record Processor-Schnittstelle](#kcl-java-v2-migration)
## Implementieren Sie die Methoden des Prozessors IRecord
Die KCL unterstützt zurzeit zwei Versionen der `IRecordProcessor`-Schnittstelle: die ursprüngliche Schnittstelle, die mit der ersten Version der KCL verfügbar war, und Version 2, die ab KCL Version 1.5.0 verfügbar ist. Beide Schnittstellen werden vollständig unterstützt. Ihre Wahl hängt von den speziellen Anforderungen Ihres Anwendungsfalls ab. Um mehr über Unterschiede zu erfahren, betrachten Sie die lokal entwickelten Javadocs oder den Quellcode. In den folgenden Abschnitten wird die Mindestimplementierung für die ersten Schritte beschrieben.
**Topics**
+ [Ursprüngliche Schnittstelle (Version 1)](#kcl-java-interface-original)
+ [Aktualisierte Schnittstelle (Version 2)](#kcl-java-interface-v2)
### Ursprüngliche Schnittstelle (Version 1)
Die ursprüngliche `IRecordProcessor` Schnittstelle (`package com.amazonaws.services.kinesis.clientlibrary.interfaces`) stellt die folgenden Datensatzverarbeitermethoden bereit, die Ihr Konsument implementieren muss. Das Beispiel stellt Implementierungen bereit, die Sie als Ausgangspunkt verwenden können (siehe `AmazonKinesisApplicationSampleRecordProcessor.java`).
```
public void initialize(String shardId)
public void processRecords(List records, IRecordProcessorCheckpointer checkpointer)
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason)
```
**initialize**
Die KCL ruft die Methode `initialize` auf, wenn der Datensatzverarbeiter instanziiert wird, und übergibt eine spezifische Shard-ID als Parameter. Dieser Datensatzverarbeiter verarbeitet nur diese Shard und in der Regel ist dies auch umgekehrt der Fall (diese Shard wird nur durch diesen Datensatverarbeiter verarbeitet). Ihr Konsument sollte jedoch die Möglichkeit berücksichtigen, dass ein Datensatz mehr als einmal verarbeitet werden könnte. Kinesis Data Streams besitzt eine Semantik nach dem Grundsatz *mindestens einmal*. Das bedeutet, dass jeder Datensatz aus einer Shard mindestens einmal von einem Worker in Ihrem Konsumenten verarbeitet wird. Weitere Informationen zu Fällen, in denen eine bestimmte Shard möglicherweise durch mehr als einen Auftragnehmer verarbeitet wird, finden Sie unter [Verwenden Sie Resharding, Skalierung und Parallelverarbeitung, um die Anzahl der Shards zu ändern](kinesis-record-processor-scaling.md).
```
public void initialize(String shardId)
```
**processRecords**
Die KCL ruft die Methode `processRecords` auf und übergibt eine Liste der Datensätze aus der Shard, die von der Methode `initialize(shardId)` angegeben wird. Der Datensatzverarbeiter verarbeitet die Daten in diesen Datensätzen entsprechend der Semantik des Konsumenten. Beispielsweise kann der Auftragnehmer eine Transformation für die Daten ausführen und das Ergebnis dann in einem Amazon Simple Storage Service (Amazon S3)-Bucket speichern.
```
public void processRecords(List records, IRecordProcessorCheckpointer checkpointer)
```
Zusätzlich zu den Daten selbst enthält der Datensatz auch eine Sequenznummer und einen Partitionsschlüssel. Der Auftragnehmer kann diese Werte beim Verarbeiten der Daten verwenden. Beispielsweise könnte der Auftragnehmer basierend auf dem Wert des Partitionsschlüssels den S3-Bucket wählen, in dem die Daten gespeichert werden sollen. Die Klasse `Record` stellt die folgenden Methoden bereit, die Zugriff auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel bieten.
```
record.getData()
record.getSequenceNumber()
record.getPartitionKey()
```
In diesem Beispiel weist die private Methode `processRecordsWithRetries` Code auf, der zeigt, wie ein Auftragnehmer auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel zugreifen kann.
Kinesis Data Streams erfordert, dass der Datensatzverarbeiter die Datensätze nachverfolgt, die bereits in einer Shard verarbeitet wurden. Die KCL übernimmt diese Nachverfolgung für Sie, indem ein Checkpointer (`IRecordProcessorCheckpointer`) an `processRecords` übergeben wird. Der Datensatzverarbeiter ruft die Methode `checkpoint` auf dieser Schnittstelle auf, um die KCL über die Fortschritte zu informieren, die sie beim Verarbeiten der Datensätze in der Shard gemacht hat. Wenn der Auftragnehmer fehlschlägt, verwendet die KCL diese Informationen, um die Verarbeitung der Shard mit dem letzten bekannten Datensatz neu zu starten.
Im Fall einer Teilungs- oder Zusammenführungsoperation beginnt die KCL erst dann mit der Verarbeitung der neuen Shards, wenn die Verarbeiter für die ursprünglichen Shards `checkpoint` aufgerufen haben, um zu signalisieren, dass die Verarbeitung der ursprünglichen Shards vollständig abgeschlossen ist.
Wenn Sie keinen Parameter übergeben, nimmt die KCL an, dass der Aufruf von `checkpoint` bedeutet, dass alle Datensätze bis zum letzten Datensatz, der an den Datensatzverarbeiter übergeben wurde, verarbeitet wurden. Daher sollte der Datensatzverarbeiter die Methode `checkpoint` erst aufrufen, wenn er alle Datensätze in der Liste, die ihm übergeben wurden, verarbeitet hat. Datensatzverarbeiter müssen `checkpoint` nicht bei jedem Aufruf von `processRecords` aufrufen. Ein Prozessor könnte beispielsweise `checkpoint` bei jedem dritten Aufruf von `processRecords` aufrufen. Sie können optional die exakte Sequenznummer eines Datensatzes als Parameter für `checkpoint` angeben. In diesem Fall nimmt die KCL an, dass alle Datensätze nur bis zu diesem Datensatz verarbeitet wurden.
Im Beispiel zeigt die private Methode `checkpoint`, wie `IRecordProcessorCheckpointer.checkpoint` mithilfe der entsprechenden Ausnahmebehandlung und Wiederholungslogik aufgerufen wird.
Die KCL ist bei der Behandlung von Ausnahmen, die während der Verarbeitung der Datensätze auftreten, von `processRecords` abhängig. Wenn `processRecords` eine Ausnahme aufwirft, überspringt die KCL die Datensätze, die vor der Ausnahme übergeben wurden. Das heißt, diese Datensätze werden nicht erneut an den Datensatzprozessor gesendet, der die Ausnahme ausgelöst hat, oder an einen anderen Datensatzprozessor im Verbraucher.
**shutdown**
Die KCL ruft die Methode `shutdown` entweder auf, wenn die Verarbeitung beendet wird (Grund für das Herunterfahren ist `TERMINATE`) oder wenn der Auftragnehmer nicht mehr reagiert (Grund für das Herunterfahren ist `ZOMBIE`).
```
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason)
```
Die Verarbeitung endet, wenn der Datensatzverarbeiter keine weiteren Datensätze aus der Shard erhält, entweder weil die Shard geteilt oder zusammengeführt wurde oder weil der Stream gelöscht wurde.
Die KCL übergibt auch eine `IRecordProcessorCheckpointer`-Schnittstelle an `shutdown`. Wenn der Grund für das Herunterfahren `TERMINATE` ist, sollte der Datensatzverarbeiter alle Datensätze fertigstellen und dann die Methode `checkpoint` in seiner Schnittstelle aufrufen.
### Aktualisierte Schnittstelle (Version 2)
Die aktualisierte `IRecordProcessor` Schnittstelle (`package com.amazonaws.services.kinesis.clientlibrary.interfaces.v2`) stellt die folgenden Datensatzverarbeitermethoden bereit, die Ihr Konsument implementieren muss:
```
void initialize(InitializationInput initializationInput)
void processRecords(ProcessRecordsInput processRecordsInput)
void shutdown(ShutdownInput shutdownInput)
```
Sie können auf alle Argumente aus der ursprünglichen Version der Schnittstelle über Get-Methoden für die Container-Objekte zugreifen. Um die Liste der Datensätze in `processRecords()` abzurufen, können Sie `processRecordsInput.getRecords()` verwenden.
Ab Version 2 dieser Schnittstelle (KCL 1.5.0 und höher) sind zusätzlich zu den Eingaben durch die ursprüngliche Schnittstelle die folgenden neuen Eingaben verfügbar:
Startsequenznummer
Im `InitializationInput`-Objekt, das an die Operation `initialize()` übergeben wird, die Startsequenznummer, aus der Datensätze für die Datenverarbeiter-Instance bereitgestellt würden. Dies ist die Sequenznummer, die zuletzt durch die Datensatzverarbeiter-Instance überprüft wurde, die dieselbe Shard zuvor verarbeitet hat. Sie wird für den Fall angegeben, dass Ihre Anwendung diese Informationen benötigt.
Ausstehende Checkpoint-Sequenznummer
Im `InitializationInput`-Objekt, das an die Operation `initialize()` übergeben wird, die ausstehende Checkpoint-Sequenznummer (wenn vorhanden), die nicht übergeben werden konnte, bevor die vorherige Datensatzverarbeiter-Instance angehalten wurde.
## Implementieren Sie eine Klassenfabrik für die IRecord Prozessorschnittstelle
Sie müssen darüber hinaus eine Factory für die Klasse implementieren, die die Datensatzverarbeitermethoden implementiert. Wenn der Konsument den Auftragnehmer instanziiert, übergibt er dieser Factory eine Referenz.
Im Beispiel wird die Factory-Klasse in der Datei `AmazonKinesisApplicationSampleRecordProcessorFactory.java` mithilfe der ursprünglichen Datensatzverarbeiter-Schnittstelle implementiert. Wenn Sie möchten, dass die Class Factory Datensatzverarbeiter mit Version 2 erstellt, verwenden Sie den Paketnamen `com.amazonaws.services.kinesis.clientlibrary.interfaces.v2`.
```
public class SampleRecordProcessorFactory implements IRecordProcessorFactory {
/**
* Constructor.
*/
public SampleRecordProcessorFactory() {
super();
}
/**
* {@inheritDoc}
*/
@Override
public IRecordProcessor createProcessor() {
return new SampleRecordProcessor();
}
}
```
## Erstellen Sie einen Worker
Wie in [Implementieren Sie die Methoden des Prozessors IRecord](#kinesis-record-processor-implementation-interface-java) beschrieben, gibt es zwei Versionen der KCL-Datensatzverarbeiterschnittstelle zur Auswahl. Die Version hat Auswirkungen auf die Art, wie Sie einen Worker erstellen. Die ursprüngliche Datensatzverarbeiterschnittstelle verwendet die folgende Codestruktur, um einen Auftragnehmer zu erstellen:
```
final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker(recordProcessorFactory, config);
```
Mit Version 2 der Datensatzverarbeiterschnittstelle können Sie `Worker.Builder` verwenden, um einen Auftragnehmer zu erstellen, ohne sich Gedanken über den Konstruktor und die Reihenfolge der Argumente zu machen. Die aktualisierte Datensatzverarbeiterschnittstelle verwendet die folgende Codestruktur, um einen Auftragnehmer zu erstellen:
```
final KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(...)
final IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory();
final Worker worker = new Worker.Builder()
.recordProcessorFactory(recordProcessorFactory)
.config(config)
.build();
```
## Ändern Sie die Konfigurationseigenschaften
Das Beispiel zeigt Standardwerte für Konfigurationseigenschaften. Diese Konfigurationsdaten für den Auftragnehmer werden anschließend in einem `KinesisClientLibConfiguration`-Objekt konsolidiert. Dieses Objekt und eine Referenz auf die Class Factory für `IRecordProcessor` werden an den Aufruf übergeben, der den Auftragnehmer instanziiert. Sie können diese Eigenschaften mithilfe einer Java-Eigenschaftendatei (siehe `AmazonKinesisApplicationSample.java`) durch eigene Werte überschreiben.
### Anwendungsname
Die KCL erfordert einen Anwendungsnamen, der unter Ihren Anwendungen sowie den Amazon-DynamoDB-Tabellen in derselben Region eindeutig ist. Sie verwendet den Wert der Anwendungsnamenkonfiguration auf folgende Arten:
+ Für mit diesem Anwendungsnamen verknüpfte Auftragnehmer wird angenommen, dass sie gemeinsam im gleichen Stream arbeiten. Diese Auftragnehmer können auf mehrere Instances verteilt sein. Wenn Sie eine zusätzliche Instance desselben Anwendungscodes ausführen, jedoch mit einem anderen Anwendungsnamen, behandelt die KCL die zweite Instance als eine völlig getrennte Anwendung, die ebenfalls im selben Stream arbeitet.
+ Die KCL erstellt eine DynamoDB-Tabelle mit dem Namen der Anwendung und verwendet die Tabelle für die Verwaltung von Statusinformationen für die Anwendung (wie Checkpoints und Auftragnehmer-Shard-Zuweisungen). Jede Anwendung verfügt über eine eigene DynamoDB-Tabelle. Weitere Informationen finden Sie unter [Verwenden Sie eine Leasetabelle, um nachzuverfolgen, welche Shards von der KCL-Consumer-Anwendung verarbeitet wurden](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable).
### Richten Sie Anmeldeinformationen ein
Sie müssen Ihre AWS Anmeldeinformationen einem der Anmeldeinformationsanbieter in der Kette der Standardanmeldedienstanbieter zur Verfügung stellen. Wenn Sie beispielsweise Ihren Konsumenten auf einer EC2-Instance ausführen, empfehlen wir, die Instance mit einer IAM-Rolle zu starten. AWS -Anmeldeinformationen, die die mit dieser IAM-Rolle verknüpften Berechtigungen widerspiegeln, werden den Anwendungen auf der Instance über deren Instance-Metadaten zur Verfügung gestellt. Dies ist die sicherste Art, Anmeldeinformationen für einen Konsumenten zu verwalten, der auf einer EC2-Instance ausgeführt wird.
Die Beispielanwendung versucht zunächst, IAM-Anmeldeinformationen aus den Instance-Metadaten abzurufen:
```
credentialsProvider = new InstanceProfileCredentialsProvider();
```
Wenn die Beispielanwendung keine Anmeldeinformationen aus den Instance-Metadaten abrufen kann, versucht sie, Anmeldeinformationen aus einer Eigenschaftendatei abzurufen:
```
credentialsProvider = new ClasspathPropertiesFileCredentialsProvider();
```
Weitere Informationen zu Instance-Metadaten finden Sie unter [Instance-Metadaten](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html) im *Amazon EC2 EC2-Benutzerhandbuch*.
### Verwenden Sie die Worker-ID für mehrere Instances
Derselbe Initialisierungscode erstellt unter Verwendung des Namens des lokalen Computers und Anfügung eines global eindeutigen Bezeichners eine ID für den Auftragnehmer, `workerId`, wie im folgenden Codeauszug gezeigt. Dieser Ansatz unterstützt das Szenario mit mehreren Instances der Konsumentenanwendung, die auf einem einzigen Computer ausgeführt werden.
```
String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID();
```
## Migrieren Sie zu Version 2 der Record Processor-Schnittstelle
Wenn Sie Code migrieren möchten, der die ursprüngliche Schnittstelle verwendet, sind zusätzlich zu den zuvor beschriebenen Schritten die folgenden Schritte erforderlich:
1. Ändern der Datensatzverarbeiterklasse, um Version 2 der Datensatzverarbeiterschnittstelle zu importieren:
```
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessor;
```
1. Ändern der Referenzen zu Eingaben, um `get`-Methoden für die Container-Objekte zu verwenden. In der Operation `shutdown()` ändern Sie beispielsweise „`checkpointer`“ in „`shutdownInput.getCheckpointer()`“.
1. Ändern der Datensatzverarbeiter-Factory, um Version 2 der Datensatzverarbeiter-Factory zu importieren:
```
import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;
```
1. Ändern der Konstruktion für den Auftragnehmer, um `Worker.Builder` zu verwenden. Beispiel:
```
final Worker worker = new Worker.Builder()
.recordProcessorFactory(recordProcessorFactory)
.config(config)
.build();
```
# Entwickeln Sie einen Kinesis Client Library-Consumer in Node.js
**Wichtig**
Die Versionen 1.x und 2.x der Amazon Kinesis Client Library (KCL) sind veraltet. KCL 1.x wird end-of-support am 30. Januar 2026 verfügbar sein. Wir **empfehlen dringend**, dass Sie Ihre KCL-Anwendungen, die Version 1.x verwenden, vor dem 30. Januar 2026 auf die neueste KCL-Version migrieren. Die neueste KCL-Version finden Sie auf der [Seite Amazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client) unter. GitHub Informationen zu den neuesten KCL-Versionen finden Sie unter. [Verwenden Sie die Kinesis-Clientbibliothek](kcl.md) Informationen zur Migration von KCL 1.x zu KCL 3.x finden Sie unter. [Migrieren von KCL 1.x zu KCL 3.x](kcl-migration-1-3.md)
Sie können die Kinesis Client Library (KCL) verwenden, um Anwendungen zu erstellen, die Daten aus Ihren Kinesis-Datenströmen verarbeiten. Die Kinesis Client Library ist in mehreren Sprachen verfügbar. In diesem Thema wird Node.js behandelt.
Die KCL ist eine Java-Bibliothek. Unterstützung für andere Sprachen als Java wird über eine mehrsprachige Schnittstelle bereitgestellt, die als. *MultiLangDaemon* Dieser Daemon basiert auf Java und wird im Hintergrund ausgeführt, wenn Sie eine andere KCL-Sprache als Java verwenden. Wenn Sie also die KCL für Node.js installieren und Ihre App für Privatanwender vollständig in Node.js schreiben, müssen Sie trotzdem Java auf Ihrem System installieren, und zwar aufgrund der. MultiLangDaemon Darüber hinaus MultiLangDaemon verfügt es über einige Standardeinstellungen, die Sie möglicherweise an Ihren Anwendungsfall anpassen müssen, z. B. die AWS Region, mit der eine Verbindung hergestellt wird. Weitere Informationen dazu finden Sie MultiLangDaemon auf GitHub der [ MultiLangDaemon KCL-Projektseite](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang).
Um die Node.js KCL von herunterzuladen GitHub, gehen Sie zur [Kinesis Client Library (Node.js)](https://github.com/awslabs/amazon-kinesis-client-nodejs).
**Downloads von Beispiel-Code**
Es gibt zwei Code-Beispiele für die KCL in Node.js:
+ [basic-sample](https://github.com/awslabs/amazon-kinesis-client-nodejs/tree/master/samples/basic_sample)
Wird in den folgenden Abschnitten verwendet, um die Grundlagen zum Erstellen einer KCL-Konsumentenanwendung in Node.js zu zeigen.
+ [click-stream-sample](https://github.com/awslabs/amazon-kinesis-client-nodejs/tree/master/samples/click_stream_sample)
Etwas komplexere und verwendet ein reales Szenario, nachdem Sie sich mit dem grundlegenden Beispiel-Code vertraut gemacht haben. Dieses Beispiel wird hier nicht behandelt. Es gibt jedoch eine Readme-Datei mit weiteren Informationen.
Sie müssen die folgenden Aufgaben durchführen, wenn Sie eine KCL-Konsumentenanwendung in Node.js implementieren:
**Topics**
+ [Implementieren Sie den Record Processor](#kinesis-record-processor-implementation-interface-nodejs)
+ [Ändern Sie die Konfigurationseigenschaften](#kinesis-record-processor-initialization-nodejs)
## Implementieren Sie den Record Processor
Der einfachste Konsument, der die KCL für Node.js verwenden kann, muss die Funktion `recordProcessor` implementieren. Diese enthält wiederum die Funktionen `initialize`, `processRecords`, und `shutdown`. Das Beispiel zeigt eine Implementierung, die Sie als Ausgangspunkt verwenden können (siehe `sample_kcl_app.js`).
```
function recordProcessor() {
// return an object that implements initialize, processRecords and shutdown functions.}
```
**initialize**
Die KCL ruft die Funktion `initialize` auf, wenn der Datensatzverarbeiter gestartet wird. Dieser Datensatzverarbeiter verarbeitet nur die Shard-ID, die als `initializeInput.shardId` übergeben wird. In der Regel ist dies auch umgekehrt der Fall (diese Shard wird nur durch diesen Datensatverarbeiter verarbeitet). Ihr Konsument sollte jedoch die Möglichkeit berücksichtigen, dass ein Datensatz mehr als einmal verarbeitet werden könnte. Das liegt daran, dass Kinesis Data Streams eine Semantik nach dem Grundsatz *mindestens einmal* hat. Das bedeutet, dass jeder Datensatz aus einer Shard mindestens einmal von einem Worker in Ihrem Konsumenten verarbeitet wird. Weitere Informationen zu Fällen, in denen eine bestimmte Shard möglicherweise durch mehr als einen Auftragnehmer verarbeitet wird, finden Sie unter [Verwenden Sie Resharding, Skalierung und Parallelverarbeitung, um die Anzahl der Shards zu ändern](kinesis-record-processor-scaling.md).
```
initialize: function(initializeInput, completeCallback)
```
**processRecords**
Die KCL ruft diese Funktion mit einer Eingabe auf, die eine Liste von Datensätzen aus der für die Funktion `initialize` angegebenen Shard enthält. Der von Ihnen implementierte Datensatzverarbeiter verarbeitet die Daten in diesen Datensätzen entsprechend der Semantik Ihres Konsumenten. Beispielsweise kann der Auftragnehmer eine Transformation für die Daten ausführen und das Ergebnis dann in einem Amazon Simple Storage Service (Amazon S3)-Bucket speichern.
```
processRecords: function(processRecordsInput, completeCallback)
```
Zusätzlich zu den Daten enthält der Datensatz auch eine Sequenznummer und einen Partitionsschlüssel, die der Auftragnehmer beim Verarbeiten der Daten verwenden kann. Beispielsweise könnte der Auftragnehmer basierend auf dem Wert des Partitionsschlüssels den S3-Bucket wählen, in dem die Daten gespeichert werden sollen. Das `record`-Anmeldeverzeichnis stellt die folgenden Schlüssel-Wert-Paare für den Zugriff auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel bereit:
```
record.data
record.sequenceNumber
record.partitionKey
```
Beachten Sie, dass die Daten Base64-kodiert sind.
Im einfachen Beispiel weist die Funktion `processRecords` Code auf, der zeigt, wie ein Auftragnehmer auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel zugreifen kann.
Kinesis Data Streams erfordert, dass der Datensatzverarbeiter die Datensätze nachverfolgt, die bereits in einer Shard verarbeitet wurden. Die KCL übernimmt die Nachverfolgung durch ein `checkpointer`-Objekt, das als `processRecordsInput.checkpointer` übergeben wird. Der Datensatzverarbeiter ruft die Funktion `checkpointer.checkpoint` auf, um die KCL über die Fortschritte zu informieren, die er beim Verarbeiten der Datensätze in der Shard gemacht hat. Wenn der Auftragnehmer fehlschlägt, verwendet die KCL diese Informationen, wenn Sie die Verarbeitung der Shard erneut starten, damit sie den Vorgang ab dem letzten bekannten Datensatz fortsetzt.
Im Fall einer Teilungs- oder Zusammenführungsoperation beginnt die KCL erst dann mit der Verarbeitung der neuen Shards, wenn die Verarbeiter für die ursprünglichen Shards `checkpoint` aufgerufen haben, um zu signalisieren, dass die Verarbeitung der ursprünglichen Shards vollständig abgeschlossen ist.
Wenn Sie die Sequenznummer nicht an die `checkpoint`-Funktion übergeben, nimmt die KCL an, dass der Aufruf von `checkpoint` bedeutet, dass alle Datensätze bis zum letzten Datensatz, der an den Datensatzverarbeiter übergeben wurde, verarbeitet wurden. Daher sollte der Datensatzverarbeiter die Methode `checkpoint` **erst** aufrufen, wenn er alle Datensätze in der Liste, die ihm übergeben wurden, verarbeitet hat. Datensatzverarbeiter müssen `checkpoint` nicht bei jedem Aufruf von `processRecords` aufrufen. Ein Prozessor könnte beispielsweise bei jedem dritten Anruf oder `checkpoint` bei einem Ereignis außerhalb Ihres Aufzeichnungsprozessors, z. B. bei einem benutzerdefinierten verification/validation Dienst, den Sie implementiert haben, aufrufen.
Sie können optional die exakte Sequenznummer eines Datensatzes als Parameter für `checkpoint` angeben. In diesem Fall nimmt die KCL an, dass alle Datensätze nur bis zu diesem Datensatz verarbeitet wurden.
Die einfache Beispielanwendung zeigt den einfachsten möglichen Aufruf der Funktion `checkpointer.checkpoint`. Sie können weitere Checkpoint-Logik hinzufügen, die Sie an diesem Punkt in der Funktion für Ihren Konsumenten benötigen.
**shutdown**
Die KCL ruft die Funktion `shutdown` entweder auf, wenn die Verarbeitung beendet wird (`shutdownInput.reason` ist `TERMINATE`) oder wenn der Auftragnehmer nicht mehr reagiert (`shutdownInput.reason` ist `ZOMBIE`).
```
shutdown: function(shutdownInput, completeCallback)
```
Die Verarbeitung endet, wenn der Datensatzverarbeiter keine weiteren Datensätze aus der Shard erhält, entweder weil die Shard geteilt oder zusammengeführt wurde oder weil der Stream gelöscht wurde.
Die KCL übergibt auch ein `shutdownInput.checkpointer`-Objekt an `shutdown`. Wenn der Grund für das Herunterfahren `TERMINATE` ist, sollten Sie sicherstellen, dass der Datensatzverarbeiter die Verarbeitung aller Datensätze fertiggestellt hat, und dann die Funktion `checkpoint` in seiner Schnittstelle aufrufen.
## Ändern Sie die Konfigurationseigenschaften
Das Beispiel zeigt Standardwerte für die Konfigurationseigenschaften. Sie können diese Eigenschaften mit eigenen Werten überschreiben (siehe `sample.properties` im einfachen Beispiel).
### Anwendungsname
Die KCL erfordert eine Anwendung, die unter Ihren Anwendungen sowie den Amazon-DynamoDB-Tabellen in derselben Region eindeutig ist. Sie verwendet den Wert der Anwendungsnamenkonfiguration auf folgende Arten:
+ Für mit diesem Anwendungsnamen verknüpfte Auftragnehmer wird angenommen, dass sie gemeinsam im gleichen Stream arbeiten. Diese Auftragnehmer können auf mehrere Instances verteilt sein. Wenn Sie eine zusätzliche Instance desselben Anwendungscodes ausführen, jedoch mit einem anderen Anwendungsnamen, behandelt die KCL die zweite Instance als eine völlig getrennte Anwendung, die ebenfalls im selben Stream arbeitet.
+ Die KCL erstellt eine DynamoDB-Tabelle mit dem Namen der Anwendung und verwendet die Tabelle für die Verwaltung von Statusinformationen für die Anwendung (wie Checkpoints und Auftragnehmer-Shard-Zuweisungen). Jede Anwendung verfügt über eine eigene DynamoDB-Tabelle. Weitere Informationen finden Sie unter [Verwenden Sie eine Leasetabelle, um nachzuverfolgen, welche Shards von der KCL-Consumer-Anwendung verarbeitet wurden](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable).
### Richten Sie Anmeldeinformationen ein
Sie müssen Ihre AWS Anmeldeinformationen einem der Anmeldeinformationsanbieter in der Kette der Standardanmeldedienstanbieter zur Verfügung stellen. Sie können die Eigenschaft `AWSCredentialsProvider` verwenden, um einen Anmeldeinformationsanbieter einzurichten. Die `sample.properties`-Datei muss Anmeldeinformationen einem der Anmeldeinformationsanbieter in der [Anmeldeinformationsanbieter-Standardkette](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html) bereitstellen. Wenn Sie Ihren Consumer auf einer Amazon EC2 EC2-Instance ausführen, empfehlen wir Ihnen, die Instance mit einer IAM-Rolle zu konfigurieren. AWS Anmeldeinformationen, die die mit dieser IAM-Rolle verknüpften Berechtigungen widerspiegeln, werden Anwendungen auf der Instance über deren Instance-Metadaten zur Verfügung gestellt. Dies ist die sicherste Art, Anmeldeinformationen für eine Konsumentenanwendung zu verwalten, die auf einer EC2-Instance ausgeführt wird.
Im folgenden Beispiel wird eine KCL konfiguriert, um einen Kinesis-Datenstrom namens `kclnodejssample` mittels des Datensatzverarbeiters zu verarbeiten, der in `sample_kcl_app.js` bereitgestellt wird:
```
# 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
```
# Entwickeln Sie einen Kinesis Client Library-Consumer in .NET
**Wichtig**
Die Versionen 1.x und 2.x der Amazon Kinesis Client Library (KCL) sind veraltet. KCL 1.x wird end-of-support am 30. Januar 2026 verfügbar sein. Wir **empfehlen dringend**, dass Sie Ihre KCL-Anwendungen, die Version 1.x verwenden, vor dem 30. Januar 2026 auf die neueste KCL-Version migrieren. Die neueste KCL-Version finden Sie auf der [Seite Amazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client) unter. GitHub Informationen zu den neuesten KCL-Versionen finden Sie unter. [Verwenden Sie die Kinesis-Clientbibliothek](kcl.md) Informationen zur Migration von KCL 1.x zu KCL 3.x finden Sie unter. [Migrieren von KCL 1.x zu KCL 3.x](kcl-migration-1-3.md)
Sie können die Kinesis Client Library (KCL) verwenden, um Anwendungen zu erstellen, die Daten aus Ihren Kinesis-Datenströmen verarbeiten. Die Kinesis Client Library ist in mehreren Sprachen verfügbar. In diesem Thema wird .NET behandelt.
Die KCL ist eine Java-Bibliothek. Unterstützung für andere Sprachen als Java wird über eine mehrsprachige Schnittstelle bereitgestellt, die als. *MultiLangDaemon* Dieser Daemon basiert auf Java und wird im Hintergrund ausgeführt, wenn Sie eine andere KCL-Sprache als Java verwenden. Wenn Sie also die KCL für .NET installieren und Ihre Consumer-App vollständig in .NET schreiben, müssen Sie trotzdem Java auf Ihrem System installieren, da MultiLangDaemon Darüber hinaus MultiLangDaemon verfügt es über einige Standardeinstellungen, die Sie möglicherweise an Ihren Anwendungsfall anpassen müssen, z. B. die AWS Region, mit der eine Verbindung hergestellt wird. Weitere Informationen dazu finden Sie MultiLangDaemon auf GitHub der [ MultiLangDaemon KCL-Projektseite](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang).
Gehen Sie zur [Kinesis Client Library (.NET) GitHub, um die.NET-KCL](https://github.com/awslabs/amazon-kinesis-client-net) von herunterzuladen. Um Beispielcode für eine.NET-KCL-Consumer-Anwendung herunterzuladen, rufen Sie die Projektseite [KCL for .NET-Beispielanwendung für Privatanwender](https://github.com/awslabs/amazon-kinesis-client-net/tree/master/SampleConsumer) auf. GitHub
Sie müssen die folgenden Aufgaben durchführen, wenn Sie eine KCL-Konsumentenanwendung in .NET implementieren:
**Topics**
+ [Implementieren Sie die Methoden der IRecord Prozessorklasse](#kinesis-record-processor-implementation-interface-dotnet)
+ [Ändern Sie die Konfigurationseigenschaften](#kinesis-record-processor-initialization-dotnet)
## Implementieren Sie die Methoden der IRecord Prozessorklasse
Der Konsument muss die folgenden Methoden für `IRecordProcessor` implementieren. Der Konsument im Beispiel stellt Implementierungen bereit, die Sie als Ausgangspunkt verwenden können (siehe die `SampleRecordProcessor`-Klasse in `SampleConsumer/AmazonKinesisSampleConsumer.cs`).
```
public void Initialize(InitializationInput input)
public void ProcessRecords(ProcessRecordsInput input)
public void Shutdown(ShutdownInput input)
```
**Initialisieren**
Die KCL ruft diese Methode auf, wenn der Datensatzverarbeiter instanziiert wird, und übergibt eine spezifische Shard-ID an den `input`-Parameter (`input.ShardId`). Dieser Datensatzverarbeiter verarbeitet nur diese Shard und in der Regel ist dies auch umgekehrt der Fall (diese Shard wird nur durch diesen Datensatverarbeiter verarbeitet). Ihr Konsument sollte jedoch die Möglichkeit berücksichtigen, dass ein Datensatz mehr als einmal verarbeitet werden könnte. Das liegt daran, dass Kinesis Data Streams eine Semantik nach dem Grundsatz *mindestens einmal* hat. Das bedeutet, dass jeder Datensatz aus einer Shard mindestens einmal von einem Worker in Ihrem Konsumenten verarbeitet wird. Weitere Informationen zu Fällen, in denen eine bestimmte Shard möglicherweise durch mehr als einen Auftragnehmer verarbeitet wird, finden Sie unter [Verwenden Sie Resharding, Skalierung und Parallelverarbeitung, um die Anzahl der Shards zu ändern](kinesis-record-processor-scaling.md).
```
public void Initialize(InitializationInput input)
```
**ProcessRecords**
Die KCL ruft diese Methode auf und übergibt eine Liste der Datensätze an den `input`-Parameter (`input.Records`) aus der Shard, die von der Methode `Initialize` angegeben wird. Der von Ihnen implementierte Datensatzverarbeiter verarbeitet die Daten in diesen Datensätzen entsprechend der Semantik Ihres Konsumenten. Beispielsweise kann der Auftragnehmer eine Transformation für die Daten ausführen und das Ergebnis dann in einem Amazon Simple Storage Service (Amazon S3)-Bucket speichern.
```
public void ProcessRecords(ProcessRecordsInput input)
```
Zusätzlich zu den Daten selbst enthält der Datensatz auch eine Sequenznummer und einen Partitionsschlüssel. Der Auftragnehmer kann diese Werte beim Verarbeiten der Daten verwenden. Beispielsweise könnte der Auftragnehmer basierend auf dem Wert des Partitionsschlüssels den S3-Bucket wählen, in dem die Daten gespeichert werden sollen. Die Klasse `Record` stellt die folgenden Methoden bereit, die Zugriff auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel bieten:
```
byte[] Record.Data
string Record.SequenceNumber
string Record.PartitionKey
```
Im Beispiel weist die Methode `ProcessRecordsWithRetries` Code auf, der zeigt, wie ein Auftragnehmer auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel zugreifen kann.
Kinesis Data Streams erfordert, dass der Datensatzverarbeiter die Datensätze nachverfolgt, die bereits in einer Shard verarbeitet wurden. Die KCL übernimmt diese Nachverfolgung für Sie, indem ein `Checkpointer`-Objekt an `ProcessRecords` (`input.Checkpointer`) übergeben wird. Der Datensatzverarbeiter ruft die Methode `Checkpointer.Checkpoint` auf, um die KCL über die Fortschritte zu informieren, die sie beim Verarbeiten der Datensätze in der Shard gemacht hat. Wenn der Auftragnehmer fehlschlägt, verwendet die KCL diese Informationen, um die Verarbeitung der Shard mit dem letzten bekannten Datensatz neu zu starten.
Im Fall einer Teilungs- oder Zusammenführungsoperation beginnt die KCL erst dann mit der Verarbeitung der neuen Shards, wenn die Verarbeiter für die ursprünglichen Shards `Checkpointer.Checkpoint` aufgerufen haben, um zu signalisieren, dass die Verarbeitung der ursprünglichen Shards vollständig abgeschlossen ist.
Wenn Sie keinen Parameter übergeben, nimmt die KCL an, dass der Aufruf von `Checkpointer.Checkpoint` bedeutet, dass alle Datensätze bis zum letzten Datensatz, der an den Datensatzverarbeiter übergeben wurde, verarbeitet wurden. Daher sollte der Datensatzverarbeiter die Methode `Checkpointer.Checkpoint` erst aufrufen, wenn er alle Datensätze in der Liste, die ihm übergeben wurden, verarbeitet hat. Datensatzverarbeiter müssen `Checkpointer.Checkpoint` nicht bei jedem Aufruf von `ProcessRecords` aufrufen. Ein Prozessor könnte beispielsweise `Checkpointer.Checkpoint` bei jedem dritten oder vierten Aufruf aufrufen. Sie können optional die exakte Sequenznummer eines Datensatzes als Parameter für `Checkpointer.Checkpoint` angeben. In diesem Fall nimmt die KCL an, dass die Datensätze nur bis zu diesem Datensatz verarbeitet wurden.
Im Beispiel zeigt die private Methode `Checkpoint(Checkpointer checkpointer)`, wie die `Checkpointer.Checkpoint`-Methode mithilfe der entsprechenden Ausnahmebehandlung und Wiederholungslogik aufgerufen wird.
Die KCL für .NET verarbeitet Ausnahmen anders als andere KCL-Sprachbibliotheken, da sie keine Ausnahmen verarbeitet, die aus der Verarbeitung der Datensätze entstanden sind. Alle nicht abgefangenen Ausnahmen vom Benutzer-Code bringen das Programm zum Absturz.
**Herunterfahren**
Die KPL ruft die Methode `Shutdown` entweder auf, wenn die Verarbeitung beendet wird (Grund für das Herunterfahren ist `TERMINATE`) oder wenn der Auftragnehmer nicht mehr reagiert (der `input.Reason`-Wert für das Herunterfahren ist `ZOMBIE`).
```
public void Shutdown(ShutdownInput input)
```
Die Verarbeitung endet, wenn der Datensatzverarbeiter keine weiteren Datensätze aus der Shard erhält, weil die Shard geteilt oder zusammengeführt wurde oder der Stream gelöscht wurde.
Die KCL übergibt auch ein `Checkpointer`-Objekt an `shutdown`. Wenn der Grund für das Herunterfahren `TERMINATE` ist, sollte der Datensatzverarbeiter alle Datensätze fertigstellen und dann die Methode `checkpoint` in seiner Schnittstelle aufrufen.
## Ändern Sie die Konfigurationseigenschaften
Der Beispielkonsument zeigt Standardwerte für die Konfigurationseigenschaften. Sie können diese Eigenschaften mit eigenen Werten überschreiben (siehe `SampleConsumer/kcl.properties`).
### Anwendungsname
Die KCL erfordert eine Anwendung, die unter Ihren Anwendungen sowie den Amazon-DynamoDB-Tabellen in derselben Region eindeutig ist. Sie verwendet den Wert der Anwendungsnamenkonfiguration auf folgende Arten:
+ Für mit diesem Anwendungsnamen verknüpfte Auftragnehmer wird angenommen, dass sie gemeinsam im gleichen Stream arbeiten. Diese Auftragnehmer können auf mehrere Instances verteilt sein. Wenn Sie eine zusätzliche Instance desselben Anwendungscodes ausführen, jedoch mit einem anderen Anwendungsnamen, behandelt die KCL die zweite Instance als eine völlig getrennte Anwendung, die ebenfalls im selben Stream arbeitet.
+ Die KCL erstellt eine DynamoDB-Tabelle mit dem Namen der Anwendung und verwendet die Tabelle für die Verwaltung von Statusinformationen für die Anwendung (wie Checkpoints und Auftragnehmer-Shard-Zuweisungen). Jede Anwendung verfügt über eine eigene DynamoDB-Tabelle. Weitere Informationen finden Sie unter [Verwenden Sie eine Leasetabelle, um nachzuverfolgen, welche Shards von der KCL-Consumer-Anwendung verarbeitet wurden](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable).
### Richten Sie Anmeldeinformationen ein
Sie müssen Ihre AWS Anmeldeinformationen einem der Anmeldeinformationsanbieter in der Kette der Standardanmeldedienstanbieter zur Verfügung stellen. Sie können die Eigenschaft `AWSCredentialsProvider` verwenden, um einen Anmeldeinformationsanbieter einzurichten. Die [sample.properties](https://github.com/awslabs/amazon-kinesis-client-python/blob/master/samples/sample.properties) muss Ihre Anmeldeinformationen einem der Anmeldeinformationsanbieter in der [Anmeldeinformationsanbieter-Standardkette](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html) bereitstellen. Wenn Sie Ihre Konsumentenanwendung auf einer EC2-Instance ausführen, empfehlen wir, die Instance mit einer IAM-Rolle zu konfigurieren. AWS -Anmeldeinformationen, die die mit dieser IAM-Rolle verknüpften Berechtigungen widerspiegeln, werden den Anwendungen auf der Instance über deren Instance-Metadaten zur Verfügung gestellt. Dies ist die sicherste Art, Anmeldeinformationen für einen Konsumenten zu verwalten, der auf einer EC2-Instance ausgeführt wird.
Die Eigenschaftendatei des Beispiels konfiguriert KCL, um einen Kinesis-Datenstrom namens „words“ mittels des Datensatzverarbeiters zu verarbeiten, der in `AmazonKinesisSampleConsumer.cs` bereitgestellt wird.
# Entwickeln Sie einen Kinesis Client Library-Consumer in Python
**Wichtig**
Die Versionen 1.x und 2.x der Amazon Kinesis Client Library (KCL) sind veraltet. KCL 1.x wird end-of-support am 30. Januar 2026 verfügbar sein. Wir **empfehlen dringend**, dass Sie Ihre KCL-Anwendungen, die Version 1.x verwenden, vor dem 30. Januar 2026 auf die neueste KCL-Version migrieren. Die neueste KCL-Version finden Sie auf der [Seite Amazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client) unter. GitHub Informationen zu den neuesten KCL-Versionen finden Sie unter. [Verwenden Sie die Kinesis-Clientbibliothek](kcl.md) Informationen zur Migration von KCL 1.x zu KCL 3.x finden Sie unter. [Migrieren von KCL 1.x zu KCL 3.x](kcl-migration-1-3.md)
Sie können die Kinesis Client Library (KCL) verwenden, um Anwendungen zu erstellen, die Daten aus Ihren Kinesis-Datenströmen verarbeiten. Die Kinesis Client Library ist in mehreren Sprachen verfügbar. In diesem Thema wird Python behandelt.
Die KCL ist eine Java-Bibliothek. Unterstützung für andere Sprachen als Java wird über eine mehrsprachige Schnittstelle bereitgestellt, die als. *MultiLangDaemon* Dieser Daemon basiert auf Java und wird im Hintergrund ausgeführt, wenn Sie eine andere KCL-Sprache als Java verwenden. Wenn Sie also die KCL für Python installieren und Ihre Consumer-App vollständig in Python schreiben, muss Java aufgrund der MultiLangDaemon trotzdem auf Ihrem System installiert sein. Darüber hinaus MultiLangDaemon verfügt es über einige Standardeinstellungen, die Sie möglicherweise an Ihren Anwendungsfall anpassen müssen, z. B. die AWS Region, mit der eine Verbindung hergestellt wird. Weitere Informationen dazu finden Sie MultiLangDaemon auf GitHub der [ MultiLangDaemon KCL-Projektseite](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang).
Um die Python-KCL von herunterzuladen GitHub, gehen Sie zur [Kinesis Client Library (Python)](https://github.com/awslabs/amazon-kinesis-client-python). Um Beispielcode für eine Python-KCL-Consumer-Anwendung herunterzuladen, gehen Sie zur [KCL for Python-Beispielprojektseite unter](https://github.com/awslabs/amazon-kinesis-client-python/tree/master/samples). GitHub
Sie müssen die folgenden Aufgaben durchführen, wenn Sie eine KCL-Konsumentenanwendung in Python implementieren:
**Topics**
+ [Implementieren Sie die Klassenmethoden RecordProcessor](#kinesis-record-processor-implementation-interface-py)
+ [Ändern Sie die Konfigurationseigenschaften](#kinesis-record-processor-initialization-py)
## Implementieren Sie die Klassenmethoden RecordProcessor
Die `RecordProcess`-Klasse muss die `RecordProcessorBase` erweitern, um die folgenden Methoden zu implementieren. Das Beispiel stellt Implementierungen bereit, die Sie als Ausgangspunkt verwenden können (siehe `sample_kclpy_app.py`).
```
def initialize(self, shard_id)
def process_records(self, records, checkpointer)
def shutdown(self, checkpointer, reason)
```
**initialize**
Die KCL ruft die Methode `initialize` auf, wenn der Datensatzverarbeiter instanziiert wird, und übergibt eine spezifische Shard-ID als Parameter. Dieser Datensatzverarbeiter verarbeitet nur diese Shard und in der Regel ist dies auch umgekehrt der Fall (diese Shard wird nur durch diesen Datensatverarbeiter verarbeitet). Ihr Konsument sollte jedoch die Möglichkeit berücksichtigen, dass ein Datensatz mehr als einmal verarbeitet werden könnte. Das liegt daran, dass Kinesis Data Streams eine Semantik nach dem Grundsatz *mindestens einmal* hat. Das bedeutet, dass jeder Datensatz aus einer Shard mindestens einmal von einem Worker in Ihrem Konsumenten verarbeitet wird. Weitere Informationen zu Fällen, in denen eine bestimmte Shard möglicherweise durch mehr als einen Auftragnehmer verarbeitet wird, finden Sie unter [Verwenden Sie Resharding, Skalierung und Parallelverarbeitung, um die Anzahl der Shards zu ändern](kinesis-record-processor-scaling.md).
```
def initialize(self, shard_id)
```
**process\$1records**
Die KCL ruft diese Methode auf und übergibt eine Liste der Datensätze aus der Shard, die von der Methode `initialize` angegeben wird. Der von Ihnen implementierte Datensatzverarbeiter verarbeitet die Daten in diesen Datensätzen entsprechend der Semantik Ihres Konsumenten. Beispielsweise kann der Auftragnehmer eine Transformation für die Daten ausführen und das Ergebnis dann in einem Amazon Simple Storage Service (Amazon S3)-Bucket speichern.
```
def process_records(self, records, checkpointer)
```
Zusätzlich zu den Daten selbst enthält der Datensatz auch eine Sequenznummer und einen Partitionsschlüssel. Der Auftragnehmer kann diese Werte beim Verarbeiten der Daten verwenden. Beispielsweise könnte der Auftragnehmer basierend auf dem Wert des Partitionsschlüssels den S3-Bucket wählen, in dem die Daten gespeichert werden sollen. Das `record`-Anmeldeverzeichnis stellt die folgenden Schlüssel-Wert-Paare für den Zugriff auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel bereit:
```
record.get('data')
record.get('sequenceNumber')
record.get('partitionKey')
```
Beachten Sie, dass die Daten Base64-kodiert sind.
Im Beispiel weist die Methode `process_records` Code auf, der zeigt, wie ein Auftragnehmer auf die Daten des Datensatzes, die Sequenznummer und den Partitionsschlüssel zugreifen kann.
Kinesis Data Streams erfordert, dass der Datensatzverarbeiter die Datensätze nachverfolgt, die bereits in einer Shard verarbeitet wurden. Die KCL übernimmt diese Nachverfolgung für Sie, indem ein `Checkpointer`-Objekt an `process_records` übergeben wird. Der Datensatzverarbeiter ruft die Methode `checkpoint` auf diesem Objekt auf, um die KCL über die Fortschritte zu informieren, die er beim Verarbeiten der Datensätze in der Shard gemacht hat. Wenn der Auftragnehmer fehlschlägt, verwendet die KCL diese Informationen, um die Verarbeitung der Shard mit dem letzten bekannten Datensatz neu zu starten.
Im Fall einer Teilungs- oder Zusammenführungsoperation beginnt die KCL erst dann mit der Verarbeitung der neuen Shards, wenn die Verarbeiter für die ursprünglichen Shards `checkpoint` aufgerufen haben, um zu signalisieren, dass die Verarbeitung der ursprünglichen Shards vollständig abgeschlossen ist.
Wenn Sie keinen Parameter übergeben, nimmt die KCL an, dass der Aufruf von `checkpoint` bedeutet, dass alle Datensätze bis zum letzten Datensatz, der an den Datensatzverarbeiter übergeben wurde, verarbeitet wurden. Daher sollte der Datensatzverarbeiter die Methode `checkpoint` erst aufrufen, wenn er alle Datensätze in der Liste, die ihm übergeben wurden, verarbeitet hat. Datensatzverarbeiter müssen `checkpoint` nicht bei jedem Aufruf von `process_records` aufrufen. Ein Prozessor könnte beispielsweise `checkpoint` bei jedem dritten Aufruf aufrufen. Sie können optional die exakte Sequenznummer eines Datensatzes als Parameter für `checkpoint` angeben. In diesem Fall nimmt die KCL an, dass alle Datensätze nur bis zu diesem Datensatz verarbeitet wurden.
Im Beispiel zeigt die private Methode `checkpoint`, wie die `Checkpointer.checkpoint`-Methode mithilfe der entsprechenden Ausnahmebehandlung und Wiederholungslogik aufgerufen wird.
Die KCL ist bei der Behandlung von Ausnahmen, die während der Verarbeitung der Datensätze auftreten, von `process_records` abhängig. Wenn `process_records` eine Ausnahme auslöst, überspringt die KCL die Datensätze, die vor der Ausnahme an `process_records` übergeben wurden. Das heißt, diese Datensätze werden nicht erneut an den Datensatzprozessor gesendet, der die Ausnahme ausgelöst hat, oder an einen anderen Datensatzprozessor im Verbraucher.
**shutdown**
Die KCL ruft die Methode `shutdown` entweder auf, wenn die Verarbeitung beendet wird (Grund für das Herunterfahren ist `TERMINATE`) oder wenn der Auftragnehmer nicht mehr reagiert (das Herunterfahren `reason` ist `ZOMBIE`).
```
def shutdown(self, checkpointer, reason)
```
Die Verarbeitung endet, wenn der Datensatzverarbeiter keine weiteren Datensätze aus der Shard erhält, entweder weil die Shard geteilt oder zusammengeführt wurde oder weil der Stream gelöscht wurde.
Die KCL übergibt auch ein `Checkpointer`-Objekt an `shutdown`. Wenn der `reason` für das Herunterfahren `TERMINATE` ist, sollte der Datensatzverarbeiter alle Datensätze fertigstellen und dann die Methode `checkpoint` in seiner Schnittstelle aufrufen.
## Ändern Sie die Konfigurationseigenschaften
Das Beispiel zeigt Standardwerte für die Konfigurationseigenschaften. Sie können diese Eigenschaften mit eigenen Werten überschreiben (siehe `sample.properties`).
### Anwendungsname
Die KCL erfordert einen Anwendungsnamen, der unter Ihren Anwendungen sowie den Amazon-DynamoDB-Tabellen in derselben Region eindeutig ist. Sie verwendet den Wert der Anwendungsnamenkonfiguration auf folgende Arten:
+ Für mit diesem Anwendungsnamen verknüpfte Auftragnehmer wird angenommen, dass sie gemeinsam im gleichen Stream arbeiten. Diese Auftragnehmer können auf mehrere Instances verteilt sein. Wenn Sie eine zusätzliche Instance desselben Anwendungscodes ausführen, jedoch mit einem anderen Anwendungsnamen, behandelt die KCL die zweite Instance als eine völlig getrennte Anwendung, die ebenfalls im selben Stream arbeitet.
+ Die KCL erstellt eine DynamoDB-Tabelle mit dem Namen der Anwendung und verwendet die Tabelle für die Verwaltung von Statusinformationen für die Anwendung (wie Checkpoints und Auftragnehmer-Shard-Zuweisungen). Jede Anwendung verfügt über eine eigene DynamoDB-Tabelle. Weitere Informationen finden Sie unter [Verwenden Sie eine Leasetabelle, um nachzuverfolgen, welche Shards von der KCL-Consumer-Anwendung verarbeitet wurden](shared-throughput-kcl-consumers.md#shared-throughput-kcl-consumers-leasetable).
### Richten Sie Anmeldeinformationen ein
Sie müssen Ihre AWS Anmeldeinformationen einem der Anmeldeinformationsanbieter in der Kette der Standardanmeldedienstanbieter zur Verfügung stellen. Sie können die Eigenschaft `AWSCredentialsProvider` verwenden, um einen Anmeldeinformationsanbieter einzurichten. Die [sample.properties](https://github.com/awslabs/amazon-kinesis-client-python/blob/master/samples/sample.properties) muss Ihre Anmeldeinformationen einem der Anmeldeinformationsanbieter in der [Anmeldeinformationsanbieter-Standardkette](https://docs.aws.amazon.com/sdk-for-java/latest/reference/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html) bereitstellen. Wenn Sie Ihre Konsumentenanwendung auf einer Amazon-EC2-Instance ausführen, empfehlen wir, die Instance mit einer IAM-Rolle zu konfigurieren. AWS -Anmeldeinformationen, die die mit dieser IAM-Rolle verknüpften Berechtigungen widerspiegeln, werden den Anwendungen auf der Instance über deren Instance-Metadaten zur Verfügung gestellt. Dies ist die sicherste Art, Anmeldeinformationen für eine Konsumentenanwendung zu verwalten, die auf einer EC2-Instance ausgeführt wird.
Die Eigenschaftendatei des Beispiels konfiguriert KCL, um einen Kinesis-Datenstrom namens „words“ mittels des Datensatzverarbeiters zu verarbeiten, der in `sample_kclpy_app.py` bereitgestellt wird.
# Entwickeln Sie einen Kinesis Client Library-Consumer in Ruby
**Wichtig**
Die Versionen 1.x und 2.x der Amazon Kinesis Client Library (KCL) sind veraltet. KCL 1.x wird end-of-support am 30. Januar 2026 verfügbar sein. Wir **empfehlen dringend**, dass Sie Ihre KCL-Anwendungen, die Version 1.x verwenden, vor dem 30. Januar 2026 auf die neueste KCL-Version migrieren. Die neueste KCL-Version finden Sie auf der [Seite Amazon Kinesis Client Library](https://github.com/awslabs/amazon-kinesis-client) unter. GitHub Informationen zu den neuesten KCL-Versionen finden Sie unter. [Verwenden Sie die Kinesis-Clientbibliothek](kcl.md) Informationen zur Migration von KCL 1.x zu KCL 3.x finden Sie unter. [Migrieren von KCL 1.x zu KCL 3.x](kcl-migration-1-3.md)
Sie können die Kinesis Client Library (KCL) verwenden, um Anwendungen zu erstellen, die Daten aus Ihren Kinesis-Datenströmen verarbeiten. Die Kinesis Client Library ist in mehreren Sprachen verfügbar. In diesem Thema wird Ruby behandelt.
Die KCL ist eine Java-Bibliothek. Unterstützung für andere Sprachen als Java wird über eine mehrsprachige Schnittstelle bereitgestellt, die als. *MultiLangDaemon* Dieser Daemon basiert auf Java und wird im Hintergrund ausgeführt, wenn Sie eine andere KCL-Sprache als Java verwenden. Wenn Sie also die KCL für Ruby installieren und Ihre Consumer-App vollständig in Ruby schreiben, müssen Sie trotzdem Java auf Ihrem System installieren, da MultiLangDaemon Darüber hinaus MultiLangDaemon verfügt es über einige Standardeinstellungen, die Sie möglicherweise an Ihren Anwendungsfall anpassen müssen, z. B. die AWS Region, mit der eine Verbindung hergestellt wird. Weitere Informationen dazu finden Sie MultiLangDaemon auf GitHub der [ MultiLangDaemon KCL-Projektseite](https://github.com/awslabs/amazon-kinesis-client/tree/v1.x/src/main/java/com/amazonaws/services/kinesis/multilang).
Um die Ruby KCL von herunterzuladen GitHub, gehen Sie zur [Kinesis Client Library (Ruby)](https://github.com/awslabs/amazon-kinesis-client-ruby). Den Beispielcode für eine Ruby-KCL-Anwendung für Privatanwender können Sie auf der Projektseite [KCL for Ruby unter](https://github.com/awslabs/amazon-kinesis-client-ruby/tree/master/samples) herunterladen. GitHub
Weitere Informationen über die KCL-Bibliothek für die Unterstützung von Ruby finden Sie unter [Dokumentation zu KCL Ruby Gems](http://www.rubydoc.info/gems/aws-kclrb).