Entwickeln Sie einen Kinesis Client Library-Consumer in Java

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 AWS Javadoc-Thema für Class. AmazonKinesisClient

Gehen Sie zur Kinesis Client Library (JavaKCL) GitHub, um Java herunterzuladen. Gehen Sie zur KCLSuchergebnisseite, um Java KCL auf Apache Maven zu finden. Um Beispielcode für eine KCL Java-Consumer-Anwendung von herunterzuladen GitHub, gehen Sie auf die Seite KCLfür das Java-Beispielprojekt unter GitHub.

Die Beispielanwendung verwendet Apache Commons Logging. 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 im Developer Guide.AWS SDK for Java

Sie müssen die folgenden Aufgaben ausführen, wenn Sie eine KCL Verbraucheranwendung in Java implementieren:

Implementieren Sie die IRecordProcessor Methoden

Die unterstützt KCL derzeit zwei Versionen der IRecordProcessor Schnittstelle: Die ursprüngliche Schnittstelle ist mit der ersten Version von verfügbarKCL, und Version 2 ist ab Version 1.5.0 verfügbar. KCL 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.

IRecordProcessorVersionen

• Ursprüngliche Schnittstelle (Version 1)
• Aktualisierte Schnittstelle (Version 2)

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<Record> records, IRecordProcessorCheckpointer checkpointer)
public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) 

initialize

Der KCL ruft die initialize Methode auf, wenn der Datensatzprozessor instanziiert wird, und übergibt eine bestimmte 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.

public void initialize(String shardId)

processRecords

Der KCL ruft die processRecords Methode auf und übergibt eine Liste von Datensätzen aus dem durch die Methode angegebenen Shard. initialize(shardId) 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<Record> 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. Der KCL kümmert sich für Sie um dieses Tracking, indem er einen Checkpointer (IRecordProcessorCheckpointer) an übergibt. processRecords Der Datensatzprozessor ruft die checkpoint Methode auf dieser Schnittstelle auf, um ihn darüber zu informieren, wie weit er bei KCL der Verarbeitung der Datensätze im Shard fortgeschritten ist. Wenn der Worker ausfällt, KCL verwendet er diese Informationen, um die Verarbeitung des Shards beim letzten bekannten verarbeiteten Datensatz neu zu starten.

Bei einem Split- oder Merge-Vorgang beginnt der KCL erst mit der Verarbeitung der neuen Shards, wenn die Prozessoren für die ursprünglichen Shards aufgerufen haben, checkpoint um zu signalisieren, dass die gesamte Verarbeitung der ursprünglichen Shards abgeschlossen ist.

Wenn Sie keinen Parameter übergeben, KCL wird davon ausgegangen, dass der Aufruf von checkpoint bedeutet, dass alle Datensätze verarbeitet wurden, bis zu dem letzten Datensatz, der an den Datensatzprozessor übergeben wurde. 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 KCL geht der davon aus, 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.

Das KCL ist darauf angewiesen, alle Ausnahmen processRecords zu behandeln, die sich aus der Verarbeitung der Datensätze ergeben. Wenn eine Ausnahme ausgelöst wirdprocessRecords, werden die Datensätze KCL übersprungen, 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 shutdown Methode wird entweder KCL aufgerufen, wenn die Verarbeitung beendet ist (der Grund für das Herunterfahren istTERMINATE) oder wenn der Worker nicht mehr reagiert (der Grund für das Herunterfahren istZOMBIE).

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.

Der übergibt KCL auch eine IRecordProcessorCheckpointer Schnittstelle anshutdown. 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 (KCL1.5.0 und höher) sind zusätzlich zu den Eingaben der ursprünglichen 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 Schnittstelle IRecordProcessor

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 Arbeiter

Wie unter beschriebenImplementieren Sie die IRecordProcessor Methoden, stehen zwei Versionen der KCL Record Processor-Schnittstelle zur Auswahl, was sich darauf auswirkt, wie Sie einen Worker erstellen würden. 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

Das KCL erfordert einen Anwendungsnamen, der für Ihre Anwendungen und für alle 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, KCL behandelt das die zweite Instance als eine völlig separate Anwendung, die ebenfalls auf demselben Stream ausgeführt wird.

• Das KCL erstellt eine DynamoDB-Tabelle mit dem Anwendungsnamen und verwendet die Tabelle, um Statusinformationen (wie Checkpoints und Worker-Shard-Mapping) für die Anwendung zu verwalten. Jede Anwendung verfügt über eine eigene DynamoDB-Tabelle. Weitere Informationen finden Sie unter Verwenden Sie eine Leasetabelle, um nachzuverfolgen, welche Shards von der Consumer-Anwendung verarbeitet wurden KCL.

Richten Sie Anmeldeinformationen ein

Sie müssen Ihre AWS Anmeldeinformationen einem der Anmeldeinformationsanbieter in der Kette der Standardanmeldedienstanbieter zur Verfügung stellen. Wenn Sie Ihren Consumer beispielsweise auf einer EC2 Instance ausführen, empfehlen wir, dass Sie die Instance mit einer IAM Rolle 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 Methode, Anmeldeinformationen für einen Verbraucher zu verwalten, der auf einer EC2 Instance läuft.

Die Beispielanwendung versucht zunächst, IAM Anmeldeinformationen aus den Metadaten der Instanz 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 im EC2Amazon-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;

2. Ä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()“.

3. Ändern der Datensatzverarbeiter-Factory, um Version 2 der Datensatzverarbeiter-Factory zu importieren:

   import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.IRecordProcessorFactory;

4. Ändern der Konstruktion für den Auftragnehmer, um Worker.Builder zu verwenden. Beispielsweise:

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