ConsumeKinesis 2.9.0

Bundle
org.apache.nifi | nifi-aws-kinesis-nar
Description
Consumes records from an Amazon Kinesis Data Stream. Uses DynamoDB-based checkpointing for reliable resumption after restarts. Note: when a shard is split or multiple shards are merged, this processor will consume from child and parent shards concurrently. It does not wait for parent shards to be fully consumed before reading child shards, so record ordering is not guaranteed across a split or merge boundary.
Tags
amazon, aws, consume, kinesis, record, stream
Input Requirement
FORBIDDEN
Supports Sensitive Dynamic Properties
false
  • Additional Details for ConsumeKinesis 2.9.0

    ConsumeKinesis

    IAM permissions required for ConsumeKinesis

    You must add the following permissions to the IAM role or user configured in AWS Credentials Provider Service.

    Cloudwatch permissions are needed only when Metrics Publishing is set to CloudWatch.

    Service Actions Resources (ARNs) Purpose
    Amazon Kinesis Data Streams DescribeStream
    DescribeStreamSummary
    RegisterStreamConsumer    		 Kinesis data stream from which ConsumeKinesis will process the data.
    arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}    		 Before attempting to read records, the consumer checks if the data stream exists, if it’s active, and if the shards are contained in the data stream. Registers consumers to a shard.
    Amazon Kinesis Data Streams GetRecords
    GetShardIterator
    ListShards    		 Kinesis data stream from which ConsumeKinesis will process the data.
    arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}    		 Reads records from a shard.
    Amazon Kinesis Data Streams SubscribeToShard
    DescribeStreamConsumer    		 Kinesis data stream from which ConsumeKinesis will process the data. Add this action only if you use enhanced fan-out (EFO) consumers.
    arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}/consumer/*    		 Subscribes to a shard for enhanced fan-out (EFO) consumers.
    Amazon DynamoDB CreateTable
    DescribeTable
    UpdateTable
    Scan
    GetItem
    PutItem
    UpdateItem
    DeleteItem    		 Lease table (metadata table in DynamoDB created by ConsumeKinesis.
    arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}    		 These actions are required for ConsumeKinesis to manage the lease table created in DynamoDB.
    Amazon DynamoDB CreateTable
    DescribeTable
    Scan
    GetItem
    PutItem
    UpdateItem
    DeleteItem    		 Worker metrics and coordinator state table (metadata tables in DynamoDB) created by ConsumeKinesis.
    arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-WorkerMetricStats
    arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-CoordinatorState    		 These actions are required for ConsumeKinesis to manage the worker metrics and coordinator state metadata tables in DynamoDB.
    Amazon DynamoDB Query Global secondary index on the lease table.
    arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}/index/*    		 This action is required for ConsumeKinesis to read the global secondary index of the lease table created in DynamoDB.
    Amazon CloudWatch PutMetricData * Upload metrics to CloudWatch that are useful for monitoring the application. The asterisk (*) is used because there is no specific resource in CloudWatch on which the PutMetricData action is invoked.

    The following is an example policy document for ConsumeKinesis.

    {
    "Version":"2012-10-17",		 	 	 
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:DescribeStreamSummary",
                "kinesis:RegisterStreamConsumer",
                "kinesis:GetRecords",
                "kinesis:GetShardIterator",
                "kinesis:ListShards"
            ],
            "Resource": [
                "arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:SubscribeToShard",
                "kinesis:DescribeStreamConsumer"
            ],
            "Resource": [
                "arn:aws:kinesis:{Region}:{Account}:stream/{Stream Name}/consumer/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:CreateTable",
                "dynamodb:DescribeTable",
                "dynamodb:UpdateTable",
                "dynamodb:GetItem",
                "dynamodb:UpdateItem",
                "dynamodb:PutItem",
                "dynamodb:DeleteItem",
                "dynamodb:Scan"
            ],
            "Resource": [
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:CreateTable",
                "dynamodb:DescribeTable",
                "dynamodb:GetItem",
                "dynamodb:UpdateItem",
                "dynamodb:PutItem",
                "dynamodb:DeleteItem",
                "dynamodb:Scan"
            ],
            "Resource": [
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-WorkerMetricStats",
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}-CoordinatorState"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:Query"
            ],
            "Resource": [
                "arn:aws:dynamodb:{Region}:{Account}:table/{Application Name}/index/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "cloudwatch:PutMetricData"
            ],
            "Resource": "*"
        }
    ]
}

    Note: Replace “{Region}”, “{Account}”, “{Stream Name}”, and “{Application Name}” in the ARNs with your own AWS region, AWS account number, Kinesis data stream name, and ConsumeKinesis Application Name property respectively.

    Consumer Type

    Comparison of different Consumer Types from Amazon Kinesis Streams documentation:

    Characteristics Shared throughput consumers without enhanced fan-out Enhanced fan-out consumers
    Read throughput Fixed at a total of 2 MB/sec per shard. If there are multiple consumers reading from the same shard, they all share this throughput. The sum of the throughputs they receive from the shard doesn’t exceed 2 MB/sec. Scales as consumers register to use enhanced fan-out. Each consumer registered to use enhanced fan-out receives its own read throughput per shard, up to 2 MB/sec, independently of other consumers.
    Message propagation delay An average of around 200 ms if you have one consumer reading from the stream. This average goes up to around 1000 ms if you have five consumers. Typically an average of 70 ms whether you have one consumer or five consumers.
    Cost Not applicable There is a data retrieval cost and a consumer-shard hour cost. For more information, see Amazon Kinesis Data Streams Pricing.

    Record processing

    When Processing Strategy property is set to RECORD, ConsumeKinesis operates in Record mode. In this mode, the processor reads records from Kinesis streams using the configured Record Reader, and writes them to FlowFiles using the configured Record Writer.

    The processor tries to optimize the number of FlowFiles created by batching multiple records with the same schema into a single FlowFile.

    Schema changes

    ConsumeKinesis supports dynamically changing Kinesis record schemas. When a record with new schema is encountered, the currently open FlowFile is closed and a new FlowFile is created for the new schema. Thanks to this, the processor preserves record ordering for a single Shard, even with record schema changes.

    Please note, the processor relies on Record Reader to provide correct schema for each record. Using a Reader with schema inference may produce a lot of different schemas, which may lead to excessive FlowFile creation. If performance is a concern, it is recommended to use a Reader with a predefined schema or schema registry.

    Output Strategies

    This processor offers multiple strategies configured via processor property Output Strategy for converting Kinesis records into FlowFiles.

    • Use Content as Value (the default) writes only a Kinesis record value to a FlowFile.
    • Use Wrapper writes a Kinesis Record value as well as metadata into separate fields of a FlowFile record.
    • Inject Metadata writes a Kinesis Record value to a FlowFile record and adds a sub-record to it with metadata.

    The written metadata includes the following fields:

    • stream: The name of the Kinesis stream the record was received from.
    • shardId: The identifier of the shard the record was received from.
    • sequenceNumber: The sequence number of the record.
    • subSequenceNumber: The subsequence number of the record, used when multiple smaller records are aggregated into a single Kinesis record. If a record was not part of a batch, this value will be 0.
    • shardedSequenceNumber: A combination of the sequence number and subsequence number. This can be used to uniquely identify a record within a shard.
    • partitionKey: The partition key of the record.
    • approximateArrival: The approximate arrival timestamp of the record (in milliseconds since epoch).

    The record schema that is used when Use Wrapper is selected is as follows (in Avro format):

    {
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [
    {
      "name": "value",
      "type": [
        {
          < Fields as determined by the Record Reader for a Kinesis message >
        },
        "null"
      ]
    },
    {
      "name": "kinesisMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "stream", "type": ["string", "null"] },
            { "name": "shardId", "type": ["string", "null"] },
            { "name": "sequenceNumber", "type": ["string", "null"] },
            { "name": "subSequenceNumber", "type": ["long", "null"] },
            { "name": "shardedSequenceNumber", "type": ["string", "null"] },
            { "name": "partitionKey", "type": ["string", "null"] },
            { "name": "approximateArrival", "type": [ { "type": "long", "logicalType": "timestamp-millis" }, "null" ] }
          ]
        },
        "null"
      ]
    }
  ]
}

    The record schema that is used when “Inject Metadata” is selected is as follows (in Avro format):

    {
  "type": "record",
  "name": "nifiRecord",
  "namespace": "org.apache.nifi",
  "fields": [
    < Fields as determined by the Record Reader for a Kinesis message >,
    {
      "name": "kinesisMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "stream", "type": ["string", "null"] },
            { "name": "shardId", "type": ["string", "null"] },
            { "name": "sequenceNumber", "type": ["string", "null"] },
            { "name": "subSequenceNumber", "type": ["long", "null"] },
            { "name": "shardedSequenceNumber", "type": ["string", "null"] },
            { "name": "partitionKey", "type": ["string", "null"] },
            { "name": "approximateArrival", "type": [ { "type": "long", "logicalType": "timestamp-millis" }, "null" ] }
          ]
        },
        "null"
      ]
    }
  ]
}

    Here is an example of FlowFile content that is emitted by JsonRecordSetWriter when strategy Use Wrapper is selected:

    [
  {
    "value": {
      "address": "1234 First Street",
      "zip": "12345",
      "account": {
        "name": "Acme",
        "number": "AC1234"
      }
    },
    "kinesisMetadata" : {
      "stream" : "stream-name",
      "shardId" : "shardId-000000000000",
      "sequenceNumber" : "123456789",
      "subSequenceNumber" : 3,
      "shardedSequenceNumber" : "12345678900000000000000000003",
      "partitionKey" : "123",
      "approximateArrival" : 1756459596788
    }
  }
]

    Here is an example of FlowFile content that is emitted by JsonRecordSetWriter when strategy Inject Metadata is selected:

    [
  {
    "address": "1234 First Street",
    "zip": "12345",
    "account": {
      "name": "Acme",
      "number": "AC1234"
    },
    "kinesisMetadata" : {
      "stream" : "stream-name",
      "shardId" : "shardId-000000000000",
      "sequenceNumber" : "123456789",
      "subSequenceNumber" : 3,
      "shardedSequenceNumber" : "12345678900000000000000000003",
      "partitionKey" : "123",
      "approximateArrival" : 1756459596788
    }
  }
]
Properties
System Resource Considerations
Resource Description
CPU The processor uses additional CPU resources when consuming data from Kinesis.
NETWORK The processor will continually poll for new Records.
MEMORY Records are fetched from Kinesis in the background and buffered in memory until they can be written to FlowFiles. Up to 200 fetch responses may be buffered per shard for both Shared Throughput and Enhanced Fan-Out. Each Shared Throughput response can contain up to the value of 'Max Records Per Request' records (default 100) and up to 10 MB, so the theoretical maximum is 20,000 records or approximately 2 GB per shard at default settings. Each Enhanced Fan-Out push event can hold up to roughly 2 MB, for a theoretical maximum of approximately 400 MB per shard. In practice the buffer is typically much smaller because fetch threads block when the queue is full and most responses are well below the maximum size.
Relationships
Name Description
success FlowFiles that are created when records are successfully read from Kinesis and parsed
Writes Attributes
Name Description
aws.kinesis.stream.name The name of the Kinesis Stream from which records were read
aws.kinesis.shard.id Shard ID from which records were read
aws.kinesis.partition.key Partition key of the last Kinesis Record in the FlowFile
aws.kinesis.first.sequence.number Sequence Number of the first Kinesis Record in the FlowFile
aws.kinesis.first.subsequence.number Sub-Sequence Number of the first Kinesis Record in the FlowFile
aws.kinesis.last.sequence.number Sequence Number of the last Kinesis Record in the FlowFile
aws.kinesis.last.subsequence.number Sub-Sequence Number of the last Kinesis Record in the FlowFile
aws.kinesis.approximate.arrival.timestamp.ms Approximate arrival timestamp associated with the Kinesis Record or records in the FlowFile
mime.type Sets the mime.type attribute to the MIME Type specified by the Record Writer (if configured)
record.count Number of records written to the FlowFile
record.error.message Error message encountered by the Record Reader or Record Writer (if configured)
kinesis.millis.behind How far behind the stream tail we are, in milliseconds