kinesis

You can use the Data Prepper kinesis source to ingest records from one or more Amazon Kinesis Data Streams.

Usage

The following example pipeline specifies Kinesis as a source. The pipeline ingests data from multiple Kinesis data streams named stream1 and stream2 and sets the initial_position to indicate the starting point for reading the stream records:

version: "2"
kinesis-pipeline:
  source:
    kinesis:
      streams:
        - stream_name: "stream1"
          initial_position: "LATEST"
        - stream_name: "stream2"
          initial_position: "LATEST"
      aws:
        region: "us-west-2"
        sts_role_arn: "arn:aws:iam::123456789012:role/my-iam-role"

Configuration options

The kinesis source supports the following configuration options.

Streams

You can use the following options in the streams array.

codec

The codec determines how the kinesis source parses each Kinesis stream record. For increased and more efficient performance, you can use codec combinations with certain processors.

json codec

The json codec parses each single line as a single JSON object from a JSON array and then creates a Data Prepper event for each object in the array. It can be used for parsing nested CloudWatch events into individual log entries. It also supports the below configuration to use with this codec.

newline codec

The newline codec parses each Kinesis stream record as a single log event, making it ideal for processing single-line records. It also works well with the parse_json processor to parse each line.

You can use the following options to configure the newline codec.

polling

When the consumer_strategy is set to polling, the kinesis source uses a polling-based approach to read records from the Kinesis data streams, instead of the default fan-out approach.

aws

You can use the following options in the aws configuration.

Exposed metadata attributes

The kinesis source adds the following metadata to each processed event. You can access the metadata attributes using the expression syntax getMetadata function.

• stream_name: Contains the name of the Kinesis data stream from which the event was obtained.

Permissions

The following minimum permissions are required in order to run kinesis as a source:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStream",
        "kinesis:DescribeStreamConsumer",
        "kinesis:DescribeStreamSummary",
        "kinesis:GetRecords",
        "kinesis:GetShardIterator",
        "kinesis:ListShards",
        "kinesis:ListStreams",
        "kinesis:ListStreamConsumers",
        "kinesis:RegisterStreamConsumer",
        "kinesis:SubscribeToShard"
      ],
      "Resource": [
        "arn:aws:kinesis:us-east-1:{account-id}:stream/stream1",
        "arn:aws:kinesis:us-east-1:{account-id}:stream/stream2"
      ]
    },
    {
      "Sid": "allowCreateTable",
      "Effect": "Allow",
      "Action": [
        "dynamodb:CreateTable",
        "dynamodb:PutItem",
        "dynamodb:DescribeTable",
        "dynamodb:DeleteItem",
        "dynamodb:GetItem",
        "dynamodb:Scan",
        "dynamodb:UpdateItem",
        "dynamodb:Query"
      ],
      "Resource": [
        "arn:aws:dynamodb:us-east-1:{account-id}:table/kinesis-pipeline"
      ]
    }
  ]
}

The kinesis source uses a DynamoDB table for ingestion coordination among multiple workers, so you need DynamoDB permissions.

Metrics

The kinesis source includes the following metrics.

Counters

• recordsProcessed: Counts the number of processed stream records.
• recordProcessingErrors: Counts the number of stream record processing errors.
• acknowledgementSetSuccesses: Counts the number of processed stream records that were successfully added to the sink.
• acknowledgementSetFailures: Counts the number of processed stream records that failed to be added to the sink.