AWSラムダ：Javaの：キネシスイベント

Question

私はトラブル私が何をしようとしているの例を見つけることを抱えているが...

私はJavaでラムダ関数を作成したいと思います。私はラムダ関数には常にJavascriptを使用すると思っていましたが、この場合はJavaで書かれたアプリケーションロジックを再利用することになりますので意味があります。私はキネシスイベントによってトリガされているJavascriptのラムダ関数を書いた過去に

。スーパーシンプルな関数は、パラメータとしてイベントを受け取り、何かを行います。私はJavaで同じことをしたいと思います。本当に簡単：

キネシス・イベント（S） - >トリガ機能 - >（Javaは）彼らと何かを

誰でも使用例このような経験を持っている、キネシスのイベントを受信しますか？ここで

Answer 1

は私が内部的に同じ概念を示すために書いたいくつかのサンプルコードです。このコードは、あるストリームから別のストリームへイベントを転送します。

注エラーが転送中である場合、このコードは再試行を処理しません。また、本番環境でのパフォーマンスであることを意味するが、それは出版ストリームからレコードを処理する方法を示しありません。

import com.amazonaws.regions.Region; 
import com.amazonaws.regions.Regions; 
import com.amazonaws.services.kinesis.AmazonKinesisClient; 
import com.amazonaws.services.kinesis.model.PutRecordsRequest; 
import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry; 
import com.amazonaws.services.kinesis.model.PutRecordsResult; 
import com.amazonaws.services.lambda.runtime.Context; 
import com.amazonaws.services.lambda.runtime.LambdaLogger; 
import com.amazonaws.services.lambda.runtime.events.KinesisEvent; 

import java.nio.ByteBuffer; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.Map; 
import java.util.stream.Collectors; 

public class KinesisToKinesis { 

    private LambdaLogger logger; 
    final private AmazonKinesisClient kinesisClient = new AmazonKinesisClient(); 

    public PutRecordsResult eventHandler(KinesisEvent event, Context context) { 
     logger = context.getLogger(); 
     if (event == null || event.getRecords() == null) { 
      logger.log("Event contains no data" + System.lineSeparator()); 
      return null; 
     } else { 
      logger.log("Received " + event.getRecords().size() + 
       " records from " + event.getRecords().get(0).getEventSourceARN() + System.lineSeparator()); 
     } 

     final Long startTime = System.currentTimeMillis(); 

     // set up the client 
     Region region; 
     final Map<String, String> environmentVariables = System.getenv(); 
     if (environmentVariables.containsKey("AWS_REGION")) { 
      region = Region.getRegion(Regions.fromName(environmentVariables.get("AWS_REGION"))); 
     } else { 
      region = Region.getRegion(Regions.US_WEST_2); 
      logger.log("Using default region: " + region.toString() + System.lineSeparator()); 
     } 
     kinesisClient.setRegion(region); 

     Long elapsed = System.currentTimeMillis() - startTime; 
     logger.log("Finished setup in " + elapsed + " ms" + System.lineSeparator()); 

     PutRecordsRequest putRecordsRequest = new PutRecordsRequest().withStreamName("usagecounters-global"); 
     List<PutRecordsRequestEntry> putRecordsRequestEntryList = event.getRecords().parallelStream() 
      .map(r -> new PutRecordsRequestEntry() 
        .withData(ByteBuffer.wrap(r.getKinesis().getData().array())) 
        .withPartitionKey(r.getKinesis().getPartitionKey())) 
      .collect(Collectors.toList()); 

     putRecordsRequest.setRecords(putRecordsRequestEntryList); 

     elapsed = System.currentTimeMillis() - startTime; 
     logger.log("Processed " + putRecordsRequest.getRecords().size() + 
      " records in " + elapsed + " ms" + System.lineSeparator()); 

     PutRecordsResult putRecordsResult = kinesisClient.putRecords(putRecordsRequest); 
     elapsed = System.currentTimeMillis() - startTime; 
     logger.log("Forwarded " + putRecordsRequest.getRecords().size() + 
       " records to Kinesis " + putRecordsRequest.getStreamName() + 
       " in " + elapsed + " ms" + System.lineSeparator()); 
     return putRecordsResult; 
    } 
}