| <!DOCTYPE html><html lang="en"><head><meta charset="utf-8"></meta><title>ConsumeKinesisStream</title><link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"></link></head><script type="text/javascript">window.onload = function(){if(self==top) { document.getElementById('nameHeader').style.display = "inherit"; } }</script><body><h1 id="nameHeader" style="display: none;">ConsumeKinesisStream</h1><h2>Description: </h2><p>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.</p><p><a href="additionalDetails.html">Additional Details...</a></p><h3>Tags: </h3><p>amazon, aws, kinesis, consume, stream</p><h3>Properties: </h3><p>In the list below, the names of required properties appear in <strong>bold</strong>. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the <a href="../../../../../html/expression-language-guide.html">NiFi Expression Language</a>.</p><table id="properties"><tr><th>Display Name</th><th>API Name</th><th>Default Value</th><th>Allowable Values</th><th>Description</th></tr><tr><td id="name"><strong>Amazon Kinesis Stream Name</strong></td><td>kinesis-stream-name</td><td></td><td id="allowable-values"></td><td id="description">The name of Kinesis Stream</td></tr><tr><td id="name"><strong>Application Name</strong></td><td>amazon-kinesis-stream-application-name</td><td></td><td id="allowable-values"></td><td id="description">The Kinesis stream reader application name.</td></tr><tr><td id="name">Record Reader</td><td>amazon-kinesis-stream-record-reader</td><td></td><td id="allowable-values"><strong>Controller Service API: </strong><br/>RecordReaderFactory<br/><strong>Implementations: </strong><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.grok.GrokReader/index.html">GrokReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.xml.XMLReader/index.html">XMLReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.avro.AvroReader/index.html">AvroReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.syslog.Syslog5424Reader/index.html">Syslog5424Reader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.cef.CEFReader/index.html">CEFReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.syslog.SyslogReader/index.html">SyslogReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.json.JsonTreeReader/index.html">JsonTreeReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.csv.CSVReader/index.html">CSVReader</a><br/><a href="../../../nifi-scripting-nar/1.19.1/org.apache.nifi.record.script.ScriptedReader/index.html">ScriptedReader</a><br/><a href="../../../nifi-parquet-nar/1.19.1/org.apache.nifi.parquet.ParquetReader/index.html">ParquetReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.json.JsonPathReader/index.html">JsonPathReader</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.lookup.ReaderLookup/index.html">ReaderLookup</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.windowsevent.WindowsEventLogReader/index.html">WindowsEventLogReader</a></td><td id="description">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.</td></tr><tr><td id="name">Record Writer</td><td>amazon-kinesis-stream-record-writer</td><td></td><td id="allowable-values"><strong>Controller Service API: </strong><br/>RecordSetWriterFactory<br/><strong>Implementations: </strong><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.text.FreeFormTextRecordSetWriter/index.html">FreeFormTextRecordSetWriter</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.avro.AvroRecordSetWriter/index.html">AvroRecordSetWriter</a><br/><a href="../../../nifi-scripting-nar/1.19.1/org.apache.nifi.record.script.ScriptedRecordSetWriter/index.html">ScriptedRecordSetWriter</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.csv.CSVRecordSetWriter/index.html">CSVRecordSetWriter</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.xml.XMLRecordSetWriter/index.html">XMLRecordSetWriter</a><br/><a href="../../../nifi-parquet-nar/1.19.1/org.apache.nifi.parquet.ParquetRecordSetWriter/index.html">ParquetRecordSetWriter</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.json.JsonRecordSetWriter/index.html">JsonRecordSetWriter</a><br/><a href="../../../nifi-record-serialization-services-nar/1.19.1/org.apache.nifi.lookup.RecordSetWriterLookup/index.html">RecordSetWriterLookup</a></td><td id="description">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.</td></tr><tr><td id="name"><strong>Region</strong></td><td>Region</td><td id="default-value">US West (Oregon)</td><td id="allowable-values"><ul><li>AWS GovCloud (US) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-gov-west-1" title="AWS Region Code : us-gov-west-1"></img></li><li>AWS GovCloud (US-East) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-gov-east-1" title="AWS Region Code : us-gov-east-1"></img></li><li>US East (N. Virginia) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-east-1" title="AWS Region Code : us-east-1"></img></li><li>US East (Ohio) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-east-2" title="AWS Region Code : us-east-2"></img></li><li>US West (N. California) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-west-1" title="AWS Region Code : us-west-1"></img></li><li>US West (Oregon) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-west-2" title="AWS Region Code : us-west-2"></img></li><li>EU (Ireland) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : eu-west-1" title="AWS Region Code : eu-west-1"></img></li><li>EU (London) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : eu-west-2" title="AWS Region Code : eu-west-2"></img></li><li>EU (Paris) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : eu-west-3" title="AWS Region Code : eu-west-3"></img></li><li>EU (Frankfurt) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : eu-central-1" title="AWS Region Code : eu-central-1"></img></li><li>EU (Stockholm) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : eu-north-1" title="AWS Region Code : eu-north-1"></img></li><li>EU (Milan) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : eu-south-1" title="AWS Region Code : eu-south-1"></img></li><li>Asia Pacific (Hong Kong) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-east-1" title="AWS Region Code : ap-east-1"></img></li><li>Asia Pacific (Mumbai) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-south-1" title="AWS Region Code : ap-south-1"></img></li><li>Asia Pacific (Singapore) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-southeast-1" title="AWS Region Code : ap-southeast-1"></img></li><li>Asia Pacific (Sydney) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-southeast-2" title="AWS Region Code : ap-southeast-2"></img></li><li>Asia Pacific (Jakarta) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-southeast-3" title="AWS Region Code : ap-southeast-3"></img></li><li>Asia Pacific (Tokyo) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-northeast-1" title="AWS Region Code : ap-northeast-1"></img></li><li>Asia Pacific (Seoul) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-northeast-2" title="AWS Region Code : ap-northeast-2"></img></li><li>Asia Pacific (Osaka) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ap-northeast-3" title="AWS Region Code : ap-northeast-3"></img></li><li>South America (Sao Paulo) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : sa-east-1" title="AWS Region Code : sa-east-1"></img></li><li>China (Beijing) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : cn-north-1" title="AWS Region Code : cn-north-1"></img></li><li>China (Ningxia) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : cn-northwest-1" title="AWS Region Code : cn-northwest-1"></img></li><li>Canada (Central) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : ca-central-1" title="AWS Region Code : ca-central-1"></img></li><li>Middle East (UAE) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : me-central-1" title="AWS Region Code : me-central-1"></img></li><li>Middle East (Bahrain) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : me-south-1" title="AWS Region Code : me-south-1"></img></li><li>Africa (Cape Town) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : af-south-1" title="AWS Region Code : af-south-1"></img></li><li>US ISO East <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-iso-east-1" title="AWS Region Code : us-iso-east-1"></img></li><li>US ISOB East (Ohio) <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-isob-east-1" title="AWS Region Code : us-isob-east-1"></img></li><li>US ISO West <img src="../../../../../html/images/iconInfo.png" alt="AWS Region Code : us-iso-west-1" title="AWS Region Code : us-iso-west-1"></img></li></ul></td><td id="description">No Description Provided.</td></tr><tr><td id="name">Endpoint Override URL</td><td>Endpoint Override URL</td><td></td><td id="allowable-values"></td><td id="description">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.<br/><strong>Supports Expression Language: true (will be evaluated using variable registry only)</strong></td></tr><tr><td id="name">DynamoDB Override</td><td>amazon-kinesis-stream-dynamodb-override</td><td></td><td id="allowable-values"></td><td id="description">DynamoDB override to use non-AWS deployments<br/><strong>Supports Expression Language: true (will be evaluated using variable registry only)</strong></td></tr><tr><td id="name"><strong>Initial Stream Position</strong></td><td>amazon-kinesis-stream-initial-position</td><td id="default-value">LATEST</td><td id="allowable-values"><ul><li>LATEST <img src="../../../../../html/images/iconInfo.png" alt="Start reading just after the most recent record in the shard, so that you always read the most recent data in the shard." title="Start reading just after the most recent record in the shard, so that you always read the most recent data in the shard."></img></li><li>TRIM_HORIZON <img src="../../../../../html/images/iconInfo.png" alt="Start reading at the last untrimmed record in the shard in the system, which is the oldest data record in the shard." title="Start reading at the last untrimmed record in the shard in the system, which is the oldest data record in the shard."></img></li><li>AT_TIMESTAMP <img src="../../../../../html/images/iconInfo.png" alt="Start reading from the position denoted by a specific time stamp, provided in the value Timestamp." title="Start reading from the position denoted by a specific time stamp, provided in the value Timestamp."></img></li></ul></td><td id="description">Initial position to read Kinesis streams.</td></tr><tr><td id="name">Stream Position Timestamp</td><td>amazon-kinesis-stream-position-timestamp</td><td></td><td id="allowable-values"></td><td id="description">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.<br/><br/><strong>This Property is only considered if </strong><strong>the [Initial Stream Position] Property has a value of "AT_TIMESTAMP".</strong></td></tr><tr><td id="name"><strong>Timestamp Format</strong></td><td>amazon-kinesis-stream-timestamp-format</td><td id="default-value">yyyy-MM-dd HH:mm:ss</td><td id="allowable-values"></td><td id="description">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.<br/><strong>Supports Expression Language: true (will be evaluated using variable registry only)</strong></td></tr><tr><td id="name"><strong>Failover Timeout</strong></td><td>amazon-kinesis-stream-failover-timeout</td><td id="default-value">30 secs</td><td id="allowable-values"></td><td id="description">Kinesis Client Library failover timeout</td></tr><tr><td id="name"><strong>Graceful Shutdown Timeout</strong></td><td>amazon-kinesis-stream-graceful-shutdown-timeout</td><td id="default-value">20 secs</td><td id="allowable-values"></td><td id="description">Kinesis Client Library graceful shutdown timeout</td></tr><tr><td id="name"><strong>Checkpoint Interval</strong></td><td>amazon-kinesis-stream-checkpoint-interval</td><td id="default-value">3 secs</td><td id="allowable-values"></td><td id="description">Interval between Kinesis checkpoints</td></tr><tr><td id="name"><strong>Retry Count</strong></td><td>amazon-kinesis-stream-retry-count</td><td id="default-value">10</td><td id="allowable-values"></td><td id="description">Number of times to retry a Kinesis operation (process record, checkpoint, shutdown)</td></tr><tr><td id="name"><strong>Retry Wait</strong></td><td>amazon-kinesis-stream-retry-wait</td><td id="default-value">1 sec</td><td id="allowable-values"></td><td id="description">Interval between Kinesis operation retries (process record, checkpoint, shutdown)</td></tr><tr><td id="name"><strong>Report Metrics to CloudWatch</strong></td><td>amazon-kinesis-stream-cloudwatch-flag</td><td id="default-value">false</td><td id="allowable-values"><ul><li>true</li><li>false</li></ul></td><td id="description">Whether to report Kinesis usage metrics to CloudWatch.</td></tr><tr><td id="name"><strong>Communications Timeout</strong></td><td>Communications Timeout</td><td id="default-value">30 secs</td><td id="allowable-values"></td><td id="description">No Description Provided.</td></tr><tr><td id="name"><strong>AWS Credentials Provider Service</strong></td><td>AWS Credentials Provider service</td><td></td><td id="allowable-values"><strong>Controller Service API: </strong><br/>AWSCredentialsProviderService<br/><strong>Implementation: </strong><a href="../org.apache.nifi.processors.aws.credentials.provider.service.AWSCredentialsProviderControllerService/index.html">AWSCredentialsProviderControllerService</a></td><td id="description">The Controller Service that is used to obtain aws credentials provider</td></tr><tr><td id="name">Proxy Configuration Service</td><td>proxy-configuration-service</td><td></td><td id="allowable-values"><strong>Controller Service API: </strong><br/>ProxyConfigurationService<br/><strong>Implementation: </strong><a href="../../../nifi-proxy-configuration-nar/1.19.1/org.apache.nifi.proxy.StandardProxyConfigurationService/index.html">StandardProxyConfigurationService</a></td><td id="description">Specifies the Proxy Configuration Controller Service to proxy network requests. If set, it supersedes proxy settings configured per component. Supported proxies: HTTP + AuthN</td></tr></table><h3>Dynamic Properties: </h3><p>Supports Sensitive Dynamic Properties: <strong>No</strong></p><p>Dynamic Properties allow the user to specify both the name and value of a property.<table id="dynamic-properties"><tr><th>Name</th><th>Value</th><th>Description</th></tr><tr><td id="name">Kinesis Client Library (KCL) Configuration property name</td><td id="value">Value to set in the KCL Configuration property</td><td>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.<br/><strong>Supports Expression Language: false</strong></td></tr></table></p><h3>Relationships: </h3><table id="relationships"><tr><th>Name</th><th>Description</th></tr><tr><td>success</td><td>FlowFiles are routed to success relationship</td></tr></table><h3>Reads Attributes: </h3>None specified.<h3>Writes Attributes: </h3><table id="writes-attributes"><tr><th>Name</th><th>Description</th></tr><tr><td>aws.kinesis.partition.key</td><td>Partition key of the (last) Kinesis Record read from the Shard</td></tr><tr><td>aws.kinesis.shard.id</td><td>Shard ID from which the Kinesis Record was read</td></tr><tr><td>aws.kinesis.sequence.number</td><td>The unique identifier of the (last) Kinesis Record within its Shard</td></tr><tr><td>aws.kinesis.approximate.arrival.timestamp</td><td>Approximate arrival timestamp of the (last) Kinesis Record read from the stream</td></tr><tr><td>mime.type</td><td>Sets the mime.type attribute to the MIME Type specified by the Record Writer (if configured)</td></tr><tr><td>record.count</td><td>Number of records written to the FlowFiles by the Record Writer (if configured)</td></tr><tr><td>record.error.message</td><td>This attribute provides on failure the error message encountered by the Record Reader or Record Writer (if configured)</td></tr></table><h3>State management: </h3>This component does not store state.<h3>Restricted: </h3>This component is not restricted.<h3>Input requirement: </h3>This component does not allow an incoming relationship.<h3>System Resource Considerations:</h3><table id="system-resource-considerations"><tr><th>Resource</th><th>Description</th></tr><tr><td>CPU</td><td>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.</td></tr><tr><td>NETWORK</td><td>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.</td></tr></table><h3>See Also:</h3><p><a href="../org.apache.nifi.processors.aws.kinesis.stream.PutKinesisStream/index.html">PutKinesisStream</a></p></body></html> |