Amazon Kinesis Firehose是否以编程方式支持数据转换?

问题描述

我有一个用例，其中我必须验证发送给Kinesis firehose的有效负载是否确实已发送.

I have a use case in which I have to verify that the payloads sent to Kinesis firehose are indeed being sent.

为此，我想到了Firehose-> Firehose Data Transformation(使用lambda)-> DDB->检查DDB中的有效负载链(有效负载是DDB中的哈希键).我必须以编程的方式一次定义整个链.数据转换与 http://docs.aws相同.amazon.com/firehose/latest/dev/data-transformation.html .

In order to do that I came up with the chain Firehose -> Firehose Data transformation(using lambda) -> DDB -> Check for payload in DDB (the payload is the hashkey in the DDB). I have to define this entire chain in one shot programatically. The data transformation is the same as http://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html.

我正在做所有这一切，因为我无法完全控制它进入的S3存储桶中的文件名.因此，我需要将确切的有效负载发送到一些持久键值存储中.

I am doing all this since I cannot completely control the file name in the S3 bucket it goes to. So I need the exact payload being sent into some persistent key value store.

问题是

http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesisfirehose/model/CreateDeliveryStreamRequest.html 似乎不支持添加数据转换lambda.

http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesisfirehose/model/CreateDeliveryStreamRequest.html does not seem to support adding a data transformation lambda.

我的问题是，这样做是否完全无需通过控制台即可完成(完全通过AWS Kinesis Firehose API).

My question is, is this doable without touching the console even once (completely through the AWS Kinesis Firehose API).

或者有其他建议以某种方式将数据移至DDB.

Or does have any alternate suggestions to move the data to DDB somehow.

推荐答案

好吧，所以我在花了很多力气和文档编写后才弄清楚了.

Well so I figured it out after much effort and documentation scrounging.

您必须使用lambda ARN定义处理配置才能定义数据转换.

You would have to define a processing configuration with a lambda ARN to define a data transform.

http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesisfirehose/model/ProcessingConfiguration.html

这是这样的配置在代码中的样子.

Here is what such a configuration would look like in code.

final ProcessingConfiguration processingConfiguration =
      new ProcessingConfiguration().withEnabled(true)
         .withProcessors(newProcessor().withType(ProcessorType.Lambda)
         .withParameters(new ProcessorParameter().withParameterName(LambdaArn)
         .withParameterValue(lamdbaFunctionArn)));

final CreateDeliveryStreamResult describeDeliveryStreamResult =
      client.createDeliveryStream(new CreateDeliveryStreamRequest().withExtendedS3DestinationConfiguration(
         new ExtendedS3DestinationConfiguration()
             .withBucketARN(s3BucketARN)
             .withRoleARN(roleArn)
             .withPrefix(keyPrefix)
             .withProcessingConfiguration(processingConfiguration))
             .withDeliveryStreamName(streamName));

此处，ARN是各种对象(S3目标，数据转换lambda，IAM角色等)的资源名称.

Here the ARNs are resource names for various objects (S3 destination, data transform lambda, IAM role etc).

这篇关于Amazon Kinesis Firehose是否以编程方式支持数据转换?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！