Amazon Kinesis

[loicmathieu]

Example scenario: consume streams from a Kinesis Data Stream, filter/manipulate data and publish into another Kinesis stream using a produce(r) task in a similar way how AWS EventBridge Pipes does that with Lambda functions https://aws.amazon.com/eventbridge/pipes/

[anna-geller]

Proposed task: PutRecords

Proposed syntax:

id: kinesis_data_streams_producer
namespace: dev
tasks:
  - id: records
    type: io.kestra.plugin.aws.kinesisdatastreams.PutRecords
    streamName: required string
    records: # required list of maps or JSON string in a structure as below:
      - data: my custom blob object e.g. "user sign-in event"
        partitionKey: required string for sharding e.g. "user1"
        explicitHashKey: optional hash value overriding the partition key
      - data: user sign-out event
        partitionKey: user1
        records2: | # only to show JSON-string payload to "records"
          [
            {
              "Data": "user sign-in event",
              "PartitionKey": "user1"
            },
            {
              "Data": "user sign-out event",
              "PartitionKey": "user1"
            }
          ]

in theory, we could offer streamArn property. However, it would lead to quite complicated logic -- most people will likely prefer streamName over streamArn. If we were to provide that as an option, we would add:

    streamArn: |
          required string if streamName is not provided. 
         There must always be either streamName or StreamArn set. 
         If this is too difficult to handle, we can provide just the "stream" property and detect whether this is ARN based on the pattern arn:aws.*:kinesis:.*:\d{12}:stream/\S+ 

Links:

• API reference
• putRecords Java SDK
• Producer developer guide with Java examples