ConsumeKinesisStream

Description:

Reads data from the specified AWS Kinesis stream and outputs a FlowFile for every processed Record (raw) or a FlowFile for a batch of processed records if a Record Reader and Record Writer are configured. At-least-once delivery of all Kinesis Records within the Stream while the processor is running. AWS Kinesis Client Library can take several seconds to initialise before starting to fetch data. Uses DynamoDB for check pointing and CloudWatch (optional) for metrics. Ensure that the credentials provided have access to DynamoDB and CloudWatch (optional) along with Kinesis.

Additional Details...

Tags:

amazon, aws, kinesis, consume, stream

Properties:

In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.

Name	Default Value	Allowable Values	Description
Amazon Kinesis Stream Name			The name of Kinesis Stream
Application Name			The Kinesis stream reader application name.
Record Reader		Controller Service API: RecordReaderFactory Implementations: JsonPathReader AvroReader XMLReader WindowsEventLogReader ReaderLookup Syslog5424Reader GrokReader ScriptedReader CSVReader SyslogReader ParquetReader JsonTreeReader CEFReader	The Record Reader to use for reading received messages. The Kinesis Stream name can be referred to by Expression Language '${kinesis.name}' to access a schema. If Record Reader/Writer are not specified, each Kinesis Record will create a FlowFile.
Record Writer		Controller Service API: RecordSetWriterFactory Implementations: AvroRecordSetWriter ScriptedRecordSetWriter JsonRecordSetWriter ParquetRecordSetWriter RecordSetWriterLookup FreeFormTextRecordSetWriter XMLRecordSetWriter CSVRecordSetWriter	The Record Writer to use for serializing Records to an output FlowFile. The Kinesis Stream name can be referred to by Expression Language '${kinesis.name}' to access a schema. If Record Reader/Writer are not specified, each Kinesis Record will create a FlowFile.
Region	us-west-2	AWS GovCloud (US) AWS GovCloud (US-East) US East (N. Virginia) US East (Ohio) US West (N. California) US West (Oregon) EU (Ireland) EU (London) EU (Paris) EU (Frankfurt) EU (Stockholm) EU (Milan) Asia Pacific (Hong Kong) Asia Pacific (Mumbai) Asia Pacific (Singapore) Asia Pacific (Sydney) Asia Pacific (Jakarta) Asia Pacific (Tokyo) Asia Pacific (Seoul) Asia Pacific (Osaka) South America (Sao Paulo) China (Beijing) China (Ningxia) Canada (Central) Middle East (Bahrain) Africa (Cape Town) US ISO East US ISOB East (Ohio) US ISO West	No Description Provided.
Endpoint Override URL			Endpoint URL to use instead of the AWS default including scheme, host, port, and path. The AWS libraries select an endpoint URL based on the AWS region, but this property overrides the selected endpoint URL, allowing use with other S3-compatible endpoints. Supports Expression Language: true (will be evaluated using variable registry only)
DynamoDB Override			DynamoDB override to use non-AWS deployments Supports Expression Language: true (will be evaluated using variable registry only)
Initial Stream Position	LATEST	LATEST TRIM_HORIZON AT_TIMESTAMP	Initial position to read Kinesis streams.
Stream Position Timestamp			Timestamp position in stream from which to start reading Kinesis Records. Required if Initial position to read Kinesis streams. is AT_TIMESTAMP. Uses the Timestamp Format to parse value into a Date. This Property is only considered if the <Initial Stream Position> Property has a value of "AT_TIMESTAMP".
Timestamp Format	yyyy-MM-dd HH:mm:ss		Format to use for parsing the Stream Position Timestamp into a Date and converting the Kinesis Record's Approximate Arrival Timestamp into a FlowFile attribute. Supports Expression Language: true (will be evaluated using variable registry only)
Failover Timeout	30 secs		Kinesis Client Library failover timeout
Graceful Shutdown Timeout	20 secs		Kinesis Client Library graceful shutdown timeout
Checkpoint Interval	3 secs		Interval between Kinesis checkpoints
Retry Count	10		Number of times to retry a Kinesis operation (process record, checkpoint, shutdown)
Retry Wait	1 sec		Interval between Kinesis operation retries (process record, checkpoint, shutdown)
Report Metrics to CloudWatch	false	true false	Whether to report Kinesis usage metrics to CloudWatch.
Communications Timeout	30 secs		No Description Provided.
AWS Credentials Provider Service		Controller Service API: AWSCredentialsProviderService Implementation: AWSCredentialsProviderControllerService	The Controller Service that is used to obtain aws credentials provider
Proxy Configuration Service		Controller Service API: ProxyConfigurationService Implementation: StandardProxyConfigurationService	Specifies the Proxy Configuration Controller Service to proxy network requests. If set, it supersedes proxy settings configured per component. Supported proxies: HTTP + AuthN

Dynamic Properties:

Dynamic Properties allow the user to specify both the name and value of a property.

Name	Value	Description
Kinesis Client Library (KCL) Configuration property name	Value to set in the KCL Configuration property	Override default KCL Configuration properties with required values. Supports setting of values via the "with" methods on the KCL Configuration class. Specify the property to be set without the leading prefix, e.g. "maxInitialisationAttempts" will call "withMaxInitialisationAttempts" and set the provided value. Only supports setting of simple property values, e.g. String, int, long and boolean. Does not allow override of KCL Configuration settings handled by non-dynamic processor properties. Supports Expression Language: false

Relationships:

Name	Description
success	FlowFiles are routed to success relationship

Reads Attributes:

None specified.

Writes Attributes:

Name	Description
aws.kinesis.partition.key	Partition key of the (last) Kinesis Record read from the Shard
aws.kinesis.shard.id	Shard ID from which the Kinesis Record was read
aws.kinesis.sequence.number	The unique identifier of the (last) Kinesis Record within its Shard
aws.kinesis.approximate.arrival.timestamp	Approximate arrival timestamp of the (last) Kinesis Record read from the stream
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 FlowFiles by the Record Writer (if configured)
record.error.message	This attribute provides on failure the error message encountered by the Record Reader or Record Writer (if configured)

State management:

This component does not store state.

Restricted:

This component is not restricted.

Input requirement:

This component does not allow an incoming relationship.

System Resource Considerations:

Resource	Description
CPU	Kinesis Client Library is used to create a Worker thread for consumption of Kinesis Records. The Worker is initialised and started when this Processor has been triggered. It runs continually, spawning Kinesis Record Processors as required to fetch Kinesis Records. The Worker Thread (and any child Record Processor threads) are not controlled by the normal NiFi scheduler as part of the Concurrent Thread pool and are not released until this processor is stopped.
NETWORK	Kinesis Client Library will continually poll for new Records, requesting up to a maximum number of Records/bytes per call. This can result in sustained network usage.

Resource

Description

CPU

Kinesis Client Library is used to create a Worker thread for consumption of Kinesis Records. The Worker is initialised and started when this Processor has been triggered. It runs continually, spawning Kinesis Record Processors as required to fetch Kinesis Records. The Worker Thread (and any child Record Processor threads) are not controlled by the normal NiFi scheduler as part of the Concurrent Thread pool and are not released until this processor is stopped.

NETWORK

Kinesis Client Library will continually poll for new Records, requesting up to a maximum number of Records/bytes per call. This can result in sustained network usage.